cdc

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

Documentation

Overview

Package cdc implements watermark management for Change Data Capture with eventual consistency.

Watermark Consistency Design

The watermark system follows a "lag-acceptable, advance-forbidden" consistency model:

  • Watermarks MAY lag behind actual data progress (causes duplicate processing, which is acceptable)
  • Watermarks MUST NEVER advance ahead of persisted data (would cause data loss, which is forbidden)

This design choice enables:

  1. Async batching for better performance (updates buffered and persisted in batches every 3s)
  2. Simplified error handling (UpdateWatermarkOnly never fails, always returns nil)
  3. Crash resilience (watermark lag on crash is acceptable, prevents data loss)

Three-Tier Cache Architecture

Watermarks flow through three cache levels before reaching the database: 

cacheUncommitted -> cacheCommitting -> cacheCommitted <-> Database

- cacheUncommitted: Immediate write buffer, updated synchronously on UpdateWatermarkOnly()
- cacheCommitting: Transition state during async batch persistence to database
- cacheCommitted: Synchronized with database, represents durable watermark state

Reads prioritize newer caches (uncommitted > committing > committed) to get latest watermark.

Failure Scenarios and Guarantees

1. System Crash Before CronJob Persists:

  • Watermarks in cacheUncommitted are lost
  • Next run reads old watermark from database
  • Result: Duplicate data processing (acceptable, handled by idempotency)

2. CronJob SQL Execution Fails:

  • Watermarks in cacheCommitting are cleared (lost)
  • Next run reads from cacheCommitted (old value) or database
  • Result: Duplicate data processing (acceptable)

3. Race Between Update and Read:

  • Reads may get stale watermark if CronJob hasn't persisted yet
  • Result: Duplicate processing (acceptable, never causes data loss)

The key guarantee: Watermarks never advance beyond successfully persisted data, ensuring no data loss even in failure scenarios.

Index

Constants

View Source 
const (
	MaxRetryCount           = 3
	ErrorExpirationDuration = 1 * time.Hour // Deprecated: No longer used
)

Error handling constants

View Source 
const (
	CDCWatermarkErrMsgMaxLen = 256

	CDCState_Running = "running"
	CDCState_Pausing = "pausing"
	CDCState_Paused  = "paused"
	CDCState_Failed  = "failed"
)
View Source 
const (
	CDCInsertTaskSqlTemplate = `INSERT INTO mo_catalog.mo_cdc_task VALUES(` +
		`%d,` +
		`"%s",` +
		`"%s",` +
		`"%s",` +
		`"%s",` +
		`"%s",` +
		`"%s",` +
		`"%s",` +
		`"%s",` +
		`"%s",` +
		`"%s",` +
		`"%s",` +
		`"%s",` +
		`"%s",` +
		`"%s",` +
		`"%s",` +
		`"%s",` +
		`"%s",` +
		`"%s",` +
		`"%s",` +
		`"%s",` +
		`%d,` +
		`"%d",` +
		`"%t",` +
		`"%s",` +
		`'%s',` +
		`"",` +
		`"",` +
		`"",` +
		`""` +
		`)`

	CDCGetTaskSqlTemplate = `SELECT ` +
		`sink_uri, ` +
		`sink_type, ` +
		`sink_password, ` +
		`tables, ` +
		`filters, ` +
		`start_ts, ` +
		`end_ts, ` +
		`no_full, ` +
		`additional_config ` +
		`FROM ` +
		`mo_catalog.mo_cdc_task ` +
		`WHERE ` +
		`account_id = %d AND ` +
		`task_id = "%s"`

	CDCShowCdcTaskSqlTemplate = `SELECT ` +
		`task_id, ` +
		`task_name, ` +
		`source_uri, ` +
		`sink_uri, ` +
		`state, ` +
		`err_msg ` +
		`FROM ` +
		`mo_catalog.mo_cdc_task ` +
		`WHERE ` +
		`account_id = %d`

	CDCGetCdcTaskIdSqlTemplate = "SELECT " +
		"task_id " +
		"FROM `mo_catalog`.`mo_cdc_task` " +
		"WHERE 1=1 AND account_id = %d"

	CDCDeleteTaskSqlTemplate = "DELETE " +
		"FROM " +
		"`mo_catalog`.`mo_cdc_task` " +
		"WHERE " +
		"1=1 AND account_id = %d"

	CDCUpdateTaskStateSqlTemplate = "UPDATE " +
		"`mo_catalog`.`mo_cdc_task` " +
		"SET state = ? " +
		"WHERE " +
		"1=1 AND account_id = %d"

	CDCUpdateTaskStateAndErrMsgSqlTemplate = "UPDATE " +
		"`mo_catalog`.`mo_cdc_task` " +
		"SET state = '%s', err_msg = '%s' " +
		"WHERE " +
		"1=1 AND account_id = %d AND task_id = '%s'"

	CDCUpdateTaskStateByTaskIdSqlTemplate = "UPDATE " +
		"`mo_catalog`.`mo_cdc_task` " +
		"SET state = '%s' " +
		"WHERE " +
		"1=1 AND account_id = %d AND task_id = '%s'"

	CDCUpdateTaskStateByTaskIdAndStateSqlTemplate = "UPDATE " +
		"`mo_catalog`.`mo_cdc_task` " +
		"SET state = '%s' " +
		"WHERE " +
		"1=1 AND account_id = %d AND task_id = '%s' AND state = '%s'"

	CDCGetDataKeySqlTemplate = "SELECT " +
		"encrypted_key " +
		"FROM mo_catalog.mo_data_key " +
		"WHERE account_id = %d and key_id = '%s'"

	// Watermark Related SQL
	CDCInsertWatermarkSqlTemplate = "INSERT INTO " +
		"`mo_catalog`.`mo_cdc_watermark` " +
		"VALUES %s"

	CDCDeleteWatermarkSqlTemplate = "DELETE FROM " +
		"`mo_catalog`.`mo_cdc_watermark` " +
		"WHERE " +
		"account_id = %d AND " +
		"task_id = '%s'"

	CDCDeleteOrphanWatermarkSqlTemplate = "DELETE w FROM " +
		"`mo_catalog`.`mo_cdc_watermark` AS w " +
		"LEFT JOIN `mo_catalog`.`mo_cdc_task` AS t " +
		"ON t.account_id = w.account_id AND t.task_id = w.task_id " +
		"WHERE w.account_id IN (%s) AND t.task_id IS NULL"

	CDCGetTableWatermarkSqlTemplate = "SELECT " +
		"watermark " +
		"FROM " +
		"`mo_catalog`.`mo_cdc_watermark` " +
		"WHERE " +
		"account_id = %d AND " +
		"task_id = '%s' AND " +
		"db_name = '%s' AND " +
		"table_name = '%s'"

	CDCGetWatermarkSqlTemplate = "SELECT " +
		"db_name, " +
		"table_name, " +
		"watermark, " +
		"err_msg " +
		"FROM " +
		"`mo_catalog`.`mo_cdc_watermark` " +
		"WHERE " +
		"account_id = %d AND " +
		"task_id = '%s'"

	CDCGetTableErrMsgSqlTemplate = "SELECT " +
		"err_msg " +
		"FROM " +
		"`mo_catalog`.`mo_cdc_watermark` " +
		"WHERE " +
		"account_id = %d AND " +
		"task_id = '%s' AND " +
		"db_name = '%s' AND " +
		"table_name = '%s'"

	CDCClearTaskTableErrorsSqlTemplate = "UPDATE " +
		"`mo_catalog`.`mo_cdc_watermark` " +
		"SET err_msg = '' " +
		"WHERE " +
		"account_id = %d AND " +
		"task_id = '%s' AND " +
		"err_msg != ''"

	CDCGetWatermarkWhereSqlTemplate = "SELECT " +
		"%s " +
		"FROM " +
		"`mo_catalog`.`mo_cdc_watermark` " +
		"WHERE %s"

	CDCOnDuplicateUpdateWatermarkTemplate = "INSERT INTO " +
		"`mo_catalog`.`mo_cdc_watermark` " +
		"(account_id, task_id, db_name, table_name, watermark) " +
		"VALUES %s " +
		"ON DUPLICATE KEY UPDATE watermark = VALUES(watermark)"

	CDCOnDuplicateUpdateWatermarkErrMsgTemplate = "INSERT INTO " +
		"`mo_catalog`.`mo_cdc_watermark` " +
		"(account_id, task_id, db_name, table_name, err_msg) " +
		"VALUES %s " +
		"ON DUPLICATE KEY UPDATE err_msg = VALUES(err_msg)"

	CDCUpdateWatermarkSqlTemplate = "UPDATE " +
		"`mo_catalog`.`mo_cdc_watermark` " +
		"SET watermark='%s' " +
		"WHERE " +
		"account_id = %d AND " +
		"task_id = '%s' AND " +
		"db_name = '%s' AND " +
		"table_name = '%s'"

	CDCDeleteWatermarkByTableSqlTemplate = "DELETE " +
		"FROM " +
		"`mo_catalog`.`mo_cdc_watermark` " +
		"WHERE " +
		"account_id = %d AND " +
		"task_id = '%s' AND " +
		"db_name = '%s' AND " +
		"table_name = '%s'"

	CDCUpdateWatermarkErrMsgSqlTemplate = "UPDATE " +
		"`mo_catalog`.`mo_cdc_watermark` " +
		"SET err_msg='%s' " +
		"WHERE " +
		"account_id = %d AND " +
		"task_id = '%s' AND " +
		"db_name = '%s' AND " +
		"table_name = '%s'"

	CDCCollectTableInfoSqlTemplate = "SELECT " +
		" rel_id, " +
		" relname, " +
		" reldatabase_id, " +
		" reldatabase, " +
		" rel_createsql, " +
		" account_id " +
		"FROM `mo_catalog`.`mo_tables` " +
		"WHERE " +
		" account_id IN (%s) " +
		"%s" +
		"%s" +
		" AND relkind = '%s' " +
		" AND reldatabase NOT IN (%s)"
	CDCInsertMOISCPLogSqlTemplate = `REPLACE INTO mo_catalog.mo_iscp_log (` +
		`account_id,` +
		`table_id,` +
		`job_name,` +
		`job_id,` +
		`job_spec,` +
		`job_state,` +
		`watermark,` +
		`job_status,` +
		`create_at,` +
		`drop_at` +
		`) VALUES (` +
		`%d,` +
		`%d,` +
		`'%s',` +
		`%d,` +
		`'%s',` +
		`'%d',` +
		`'%s',` +
		`'%s',` +
		`now(),` +
		`null` +
		`)`
	CDCUpdateMOISCPLogSqlTemplate = `UPDATE mo_catalog.mo_iscp_log SET ` +
		`job_state = %d,` +
		`watermark = '%s',` +
		`job_status = '%s'` +
		`WHERE` +
		` account_id = %d ` +
		`AND table_id = %d ` +
		`AND job_name = '%s'` +
		`AND job_id = %d ` +
		`AND job_state != 4 ` +
		`AND  JSON_EXTRACT(job_status, '$.LSN') = '%d'`
	CDCUpdateMOISCPLogJobSpecSqlTemplate = `UPDATE mo_catalog.mo_iscp_log SET ` +
		`job_spec = '%s'` +
		`WHERE` +
		` account_id = %d ` +
		`AND table_id = %d ` +
		`AND job_name = '%s'` +
		`AND job_id = %d`
	CDCUpdateMOISCPLogDropAtSqlTemplate = `UPDATE mo_catalog.mo_iscp_log SET ` +
		`drop_at = now()` +
		`WHERE` +
		` account_id = %d ` +
		`AND table_id = %d ` +
		`AND job_name = '%s'` +
		`AND job_id = %d`
	CDCDeleteMOISCPLogSqlTemplate = `DELETE FROM mo_catalog.mo_iscp_log WHERE ` +
		`drop_at < '%s'`
	CDCSelectMOISCPLogSqlTemplate        = `SELECT * from mo_catalog.mo_iscp_log`
	CDCSelectMOISCPLogByTableSqlTemplate = `SELECT drop_at, job_id from mo_catalog.mo_iscp_log WHERE ` +
		`account_id = %d ` +
		`AND table_id = %d ` +
		`AND job_name = '%s'`
	CDCGetTableIDTemplate = "SELECT " +
		"rel_id, " +
		"reldatabase_id " +
		"FROM `mo_catalog`.`mo_tables` " +
		"WHERE " +
		" account_id = %d " +
		" AND reldatabase = '%s' " +
		" AND relname = '%s' "
)
View Source 
const (
	CDCInsertTaskSqlTemplate_Idx                    = 0
	CDCGetTaskSqlTemplate_Idx                       = 1
	CDCShowTaskSqlTemplate_Idx                      = 2
	CDCGetTaskIdSqlTemplate_Idx                     = 3
	CDCDeleteTaskSqlTemplate_Idx                    = 4
	CDCUpdateTaskStateSQL_Idx                       = 5
	CDCUpdateTaskStateAndErrMsgSQL_Idx              = 6
	CDCInsertWatermarkSqlTemplate_Idx               = 7
	CDCGetWatermarkSqlTemplate_Idx                  = 8
	CDCGetTableWatermarkSQL_Idx                     = 9
	CDCGetTableErrMsgSQL_Idx                        = 10
	CDCUpdateWatermarkSQL_Idx                       = 11
	CDCUpdateWatermarkErrMsgSQL_Idx                 = 12
	CDCDeleteWatermarkSqlTemplate_Idx               = 13
	CDCDeleteWatermarkByTableSqlTemplate_Idx        = 14
	CDCDeleteOrphanWatermarkSqlTemplate_Idx         = 15
	CDCGetDataKeySQL_Idx                            = 16
	CDCCollectTableInfoSqlTemplate_Idx              = 17
	CDCGetWatermarkWhereSqlTemplate_Idx             = 18
	CDCOnDuplicateUpdateWatermarkTemplate_Idx       = 19
	CDCOnDuplicateUpdateWatermarkErrMsgTemplate_Idx = 20
	CDCClearTaskTableErrorsSQL_Idx                  = 21
	CDCInsertMOISCPLogSqlTemplate_Idx               = 22
	CDCUpdateMOISCPLogSqlTemplate_Idx               = 23
	CDCUpdateMOISCPLogDropAtSqlTemplate_Idx         = 24
	CDCDeleteMOISCPLogSqlTemplate_Idx               = 25
	CDCSelectMOISCPLogSqlTemplate_Idx               = 26
	CDCSelectMOISCPLogByTableSqlTemplate_Idx        = 27
	CDCUpdateMOISCPLogJobSpecSqlTemplate_Idx        = 28
	CDCGetTableIDTemplate_Idx                       = 29
	CDCUpdateTaskStateByTaskIdSQL_Idx               = 30
	CDCUpdateTaskStateByTaskIdAndStateSQL_Idx       = 31

	CDCSqlTemplateCount = 32
)
View Source 
const (
	DefaultSlowThreshold          = 10 * time.Minute
	DefaultPrintInterval          = 5 * time.Minute
	DefaultWatermarkCleanupPeriod = 5 * time.Minute
	DefaultCleanupWarnThreshold   = 5 * time.Second
)
View Source 
const (
	DefaultFrequency     = 200 * time.Millisecond
	RetryableErrorPrefix = "retryable error:"
)

Reader lifecycle constants

View Source 
const (
	CDCSourceUriPrefix = "mysql://"
	CDCSinkUriPrefix   = "mysql://"

	CDCState_Common = "common"
	CDCState_Error  = "error"
)
View Source 
const (
	CDCDefaultSendSqlTimeout                 = "10m"
	CDCDefaultRetryTimes                     = -1
	CDCDefaultRetryDuration                  = 10 * time.Minute
	CDCDefaultTaskExtra_InitSnapshotSplitTxn = true
	CDCDefaultTaskExtra_MaxSQLLen            = 4 * 1024 * 1024
)
View Source 
const (
	CDCSinkType_MySQL   = "mysql"
	CDCSinkType_MO      = "matrixone"
	CDCSinkType_Console = "console"
)
View Source 
const (
	CDCPitrGranularity_Table   = "table"
	CDCPitrGranularity_DB      = "database"
	CDCPitrGranularity_Account = "account"
	CDCPitrGranularity_Cluster = "cluster"
	CDCPitrGranularity_All     = "*"
)
View Source 
const (
	CDCRequestOptions_Level                = "Level"
	CDCRequestOptions_Exclude              = "Exclude"
	CDCRequestOptions_StartTs              = "StartTs"
	CDCRequestOptions_EndTs                = "EndTs"
	CDCRequestOptions_SendSqlTimeout       = "SendSqlTimeout"
	CDCRequestOptions_InitSnapshotSplitTxn = "InitSnapshotSplitTxn"
	CDCRequestOptions_MaxSqlLength         = "MaxSqlLength"
	CDCRequestOptions_NoFull               = "NoFull"
	CDCRequestOptions_ConfigFile           = "ConfigFile"
	CDCRequestOptions_Frequency            = "Frequency"
)
View Source 
const (
	CDCTaskExtraOptions_MaxSqlLength         = CDCRequestOptions_MaxSqlLength
	CDCTaskExtraOptions_SendSqlTimeout       = CDCRequestOptions_SendSqlTimeout
	CDCTaskExtraOptions_InitSnapshotSplitTxn = CDCRequestOptions_InitSnapshotSplitTxn
	CDCTaskExtraOptions_Frequency            = CDCRequestOptions_Frequency
)
View Source 
const (
	InitKeyId           = "4e3da275-5003-4ca0-8667-5d3cdbecdd35"
	InsertDataKeyFormat = "replace into mo_catalog.mo_data_key (account_id, key_id, encrypted_key) values (%d, '%s', '%s')"
)
View Source 
const (
	WatermarkUpdateInterval          = time.Second * 3
	ReadWatermarkProjectionList      = "account_id, task_id, db_name, table_name, watermark"
	UpdateWatermarkCronJobNamePrefix = "CDCWatermarkUpdater-CronJob"
)
View Source 
const (
	JT_CDC_GetOrAddCommittedWM tasks.JobType = 400 + iota
	JT_CDC_CommittingWM
	JT_CDC_UpdateWMErrMsg
	JT_CDC_RemoveCachedWM
)

Variables

View Source 
var AesCFBDecodeWithKey = func(ctx context.Context, data string, aesKey []byte) (string, error) {
	if len(aesKey) == 0 {
		return "", moerr.NewInternalErrorNoCtx("AesKey is not initialized")
	}

	encodedData, err := hex.DecodeString(data)
	if err != nil {
		return "", err
	}
	block, err := aes.NewCipher(aesKey)
	if err != nil {
		return "", err
	}
	if len(encodedData) < aes.BlockSize {
		return "", moerr.NewInternalError(ctx, "encoded string is too short")
	}
	iv := encodedData[:aes.BlockSize]
	encodedData = encodedData[aes.BlockSize:]
	stream := cipher.NewCTR(block, iv)
	stream.XORKeyStream(encodedData, encodedData)
	return string(encodedData), nil
}
View Source 
var AesKey string
View Source 
var CDCRequestOptions = []string{
	CDCRequestOptions_Level,
	CDCRequestOptions_Exclude,
	CDCRequestOptions_StartTs,
	CDCRequestOptions_EndTs,
	CDCRequestOptions_MaxSqlLength,
	CDCRequestOptions_SendSqlTimeout,
	CDCRequestOptions_InitSnapshotSplitTxn,
	CDCRequestOptions_ConfigFile,
	CDCRequestOptions_NoFull,
	CDCRequestOptions_Frequency,
}
View Source 
var CDCSQLBuilder = cdcSQLBuilder{}
View Source 
var CDCSQLTemplates = [CDCSqlTemplateCount]struct {
	SQL         string
	OutputAttrs []string
}{
	CDCInsertTaskSqlTemplate_Idx: {
		SQL: CDCInsertTaskSqlTemplate,
	},
	CDCGetTaskSqlTemplate_Idx: {
		SQL: CDCGetTaskSqlTemplate,
		OutputAttrs: []string{
			"sink_uri",
			"sink_type",
			"sink_password",
			"tables",
			"filters",
			"start_ts",
			"end_ts",
			"no_full",
			"additional_config",
		},
	},
	CDCShowTaskSqlTemplate_Idx: {
		SQL: CDCShowCdcTaskSqlTemplate,
		OutputAttrs: []string{
			"task_id",
			"task_name",
			"source_uri",
			"sink_uri",
			"state",
			"err_msg",
		},
	},
	CDCGetTaskIdSqlTemplate_Idx: {
		SQL:         CDCGetCdcTaskIdSqlTemplate,
		OutputAttrs: []string{"task_id"},
	},
	CDCDeleteTaskSqlTemplate_Idx: {
		SQL: CDCDeleteTaskSqlTemplate,
	},
	CDCUpdateTaskStateSQL_Idx: {
		SQL: CDCUpdateTaskStateSqlTemplate,
	},
	CDCUpdateTaskStateAndErrMsgSQL_Idx: {
		SQL: CDCUpdateTaskStateAndErrMsgSqlTemplate,
	},
	CDCUpdateTaskStateByTaskIdSQL_Idx: {
		SQL: CDCUpdateTaskStateByTaskIdSqlTemplate,
	},
	CDCUpdateTaskStateByTaskIdAndStateSQL_Idx: {
		SQL: CDCUpdateTaskStateByTaskIdAndStateSqlTemplate,
	},
	CDCInsertWatermarkSqlTemplate_Idx: {
		SQL: CDCInsertWatermarkSqlTemplate,
	},
	CDCGetWatermarkSqlTemplate_Idx: {
		SQL: CDCGetWatermarkSqlTemplate,
		OutputAttrs: []string{
			"db_name",
			"table_name",
			"watermark",
			"err_msg",
		},
	},
	CDCGetTableWatermarkSQL_Idx: {
		SQL: CDCGetTableWatermarkSqlTemplate,
		OutputAttrs: []string{
			"watermark",
		},
	},
	CDCGetTableErrMsgSQL_Idx: {
		SQL: CDCGetTableErrMsgSqlTemplate,
		OutputAttrs: []string{
			"err_msg",
		},
	},
	CDCUpdateWatermarkSQL_Idx: {
		SQL: CDCUpdateWatermarkSqlTemplate,
	},
	CDCUpdateWatermarkErrMsgSQL_Idx: {
		SQL: CDCUpdateWatermarkErrMsgSqlTemplate,
	},
	CDCDeleteWatermarkSqlTemplate_Idx: {
		SQL: CDCDeleteWatermarkSqlTemplate,
	},
	CDCDeleteOrphanWatermarkSqlTemplate_Idx: {
		SQL: CDCDeleteOrphanWatermarkSqlTemplate,
	},
	CDCDeleteWatermarkByTableSqlTemplate_Idx: {
		SQL: CDCDeleteWatermarkByTableSqlTemplate,
	},
	CDCGetDataKeySQL_Idx: {
		SQL:         CDCGetDataKeySqlTemplate,
		OutputAttrs: []string{"encrypted_key"},
	},
	CDCCollectTableInfoSqlTemplate_Idx: {
		SQL: CDCCollectTableInfoSqlTemplate,
		OutputAttrs: []string{
			"rel_id",
			"relname",
			"reldatabase_id",
			"reldatabase",
			"rel_createsql",
			"account_id",
		},
	},
	CDCGetWatermarkWhereSqlTemplate_Idx: {
		SQL: CDCGetWatermarkWhereSqlTemplate,
	},
	CDCOnDuplicateUpdateWatermarkTemplate_Idx: {
		SQL: CDCOnDuplicateUpdateWatermarkTemplate,
	},
	CDCOnDuplicateUpdateWatermarkErrMsgTemplate_Idx: {
		SQL: CDCOnDuplicateUpdateWatermarkErrMsgTemplate,
	},
	CDCClearTaskTableErrorsSQL_Idx: {
		SQL: CDCClearTaskTableErrorsSqlTemplate,
	},
	CDCInsertMOISCPLogSqlTemplate_Idx: {
		SQL: CDCInsertMOISCPLogSqlTemplate,
	},
	CDCUpdateMOISCPLogSqlTemplate_Idx: {
		SQL: CDCUpdateMOISCPLogSqlTemplate,
	},
	CDCUpdateMOISCPLogDropAtSqlTemplate_Idx: {
		SQL: CDCUpdateMOISCPLogDropAtSqlTemplate,
	},
	CDCDeleteMOISCPLogSqlTemplate_Idx: {
		SQL: CDCDeleteMOISCPLogSqlTemplate,
	},
	CDCSelectMOISCPLogSqlTemplate_Idx: {
		SQL: CDCSelectMOISCPLogSqlTemplate,
		OutputAttrs: []string{
			"account_id",
			"table_id",
			"job_name",
			"job_id",
			"job_spec",
			"job_state",
			"watermark",
			"job_status",
			"create_at",
			"drop_at",
		},
	},
	CDCSelectMOISCPLogByTableSqlTemplate_Idx: {
		SQL: CDCSelectMOISCPLogByTableSqlTemplate,
		OutputAttrs: []string{
			"drop_at",
			"job_id",
		},
	},
	CDCUpdateMOISCPLogJobSpecSqlTemplate_Idx: {
		SQL: CDCUpdateMOISCPLogJobSpecSqlTemplate,
	},
	CDCGetTableIDTemplate_Idx: {
		SQL: CDCGetTableIDTemplate,
		OutputAttrs: []string{
			"rel_id",
			"reldatabase_id",
		},
	},
}
View Source 
var CDCTaskExtraOptions = []string{
	CDCTaskExtraOptions_MaxSqlLength,
	CDCTaskExtraOptions_SendSqlTimeout,
	CDCTaskExtraOptions_InitSnapshotSplitTxn,
	CDCTaskExtraOptions_Frequency,
}
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, true, mp)
}
View Source 
var CreateMysqlSinker2 = func(
	sinkUri UriInfo,
	accountId uint64,
	taskId string,
	dbTblInfo *DbTableInfo,
	watermarkUpdater *CDCWatermarkUpdater,
	tableDef *plan.TableDef,
	retryTimes int,
	retryDuration time.Duration,
	ar *ActiveRoutine,
	maxSqlLength uint64,
	sendSqlTimeout string,
) (Sinker, error) {
	// 1. Determine if we need to record transactions for debugging
	var doRecord bool
	if tableDef != nil {
		doRecord, _ = objectio.CDCRecordTxnInjected(tableDef.DbName, tableDef.Name)
	}

	executor, err := NewExecutor(
		sinkUri.User, sinkUri.Password,
		sinkUri.Ip, sinkUri.Port,
		retryTimes, retryDuration,
		sendSqlTimeout,
		doRecord,
	)
	if err != nil {
		return nil, err
	}

	ctx := context.Background()

	addPadding := func(sql string) []byte {
		padding := strings.Repeat(" ", v2SQLBufReserved)
		return []byte(padding + sql)
	}

	createDbSQL := fmt.Sprintf("CREATE DATABASE IF NOT EXISTS `%s`", dbTblInfo.SinkDbName)
	err = executor.ExecSQL(ctx, ar, addPadding(createDbSQL), false)
	if err != nil {
		executor.Close()
		return nil, err
	}

	useDbSQL := fmt.Sprintf("USE `%s`", dbTblInfo.SinkDbName)
	err = executor.ExecSQL(ctx, ar, addPadding(useDbSQL), false)
	if err != nil {
		executor.Close()
		return nil, err
	}

	if dbTblInfo.IdChanged {
		dropTableSQL := fmt.Sprintf("DROP TABLE IF EXISTS `%s`", dbTblInfo.SinkTblName)
		err = executor.ExecSQL(ctx, ar, addPadding(dropTableSQL), false)
		if err != nil {
			executor.Close()
			return nil, err
		}
		dbTblInfo.IdChanged = false
	}

	// CREATE TABLE
	var createSql string
	if tableDef != nil {
		newTableDef := *tableDef
		newTableDef.DbName = dbTblInfo.SinkDbName
		newTableDef.Name = dbTblInfo.SinkTblName
		newTableDef.Fkeys = nil
		newTableDef.Partition = nil

		if newTableDef.TableType == catalog.SystemClusterRel {
			executor.Close()
			return nil, moerr.NewInternalErrorNoCtx("cluster table is not supported")
		}
		if newTableDef.TableType == catalog.SystemExternalRel {
			executor.Close()
			return nil, moerr.NewInternalErrorNoCtx("external table is not supported")
		}

		createSql, _, err = plan2.ConstructCreateTableSQL(nil, &newTableDef, nil, true, nil)
		if err != nil {
			executor.Close()
			return nil, err
		}
		createSql = strings.Replace(createSql, "CREATE TABLE", "CREATE TABLE IF NOT EXISTS", 1)
	}

	err = executor.ExecSQL(ctx, ar, addPadding(createSql), false)
	if err != nil {
		executor.Close()
		return nil, err
	}

	builder, err := NewCDCStatementBuilder(
		dbTblInfo.SinkDbName,
		dbTblInfo.SinkTblName,
		tableDef,
		maxSqlLength,
		sinkUri.SinkTyp == CDCSinkType_MO,
	)
	if err != nil {
		executor.Close()
		return nil, err
	}

	sinker := NewMysqlSinker2(
		executor,
		accountId,
		taskId,
		dbTblInfo,
		watermarkUpdater,
		builder,
		ar,
	)

	logutil.Info(
		"cdc.mysql_sinker2.create_success",
		zap.String("db", dbTblInfo.SinkDbName),
		zap.String("table", dbTblInfo.SinkTblName),
		zap.String("task-id", taskId),
		zap.Uint64("account-id", accountId),
	)

	return sinker, nil
}

CreateMysqlSinker2 creates a new MySQL sinker with v2 architecture This is the main entry point for creating CDC sink, replacing the old CreateMysqlSinker Defined as var for test mocking

View Source 
var (
	EnableConsoleSink = false
)
View Source 
var EnterRunSql = func(ctx context.Context, txnOp client.TxnOperator, sql string) func() {
	if txnOp == nil {
		return func() {}
	}
	if ctx == nil {
		ctx = context.Background()
	}
	_, cancel := context.WithCancel(ctx)
	token := txnOp.EnterRunSqlWithTokenAndSQL(cancel, sql)
	return func() {
		txnOp.ExitRunSqlWithToken(token)
		cancel()
	}
}
View Source 
var ErrNoWatermarkFound = moerr.NewInternalErrorNoCtx("no watermark found")
View Source 
var ErrSetAlreadyPersisted = moerr.NewInternalErrorNoCtx("set already persisted")
View Source 
var FinishTxnOp = func(ctx context.Context, inputErr error, txnOp client.TxnOperator, cnEngine engine.Engine) {

	ctx2, cancel := context.WithTimeoutCause(ctx, cnEngine.Hints().CommitOrRollbackTimeout, moerr.CauseFinishTxnOp)
	defer cancel()
	if inputErr != nil {
		_ = txnOp.Rollback(ctx2)
	} else {
		_ = txnOp.Commit(ctx2)
	}
}
View Source 
var GetRelationById = func(ctx context.Context, cnEngine engine.Engine, txnOp client.TxnOperator, tableId uint64) (dbName string, tblName string, rel engine.Relation, err error) {
	return cnEngine.GetRelationById(ctx, txnOp, tableId)
}
View Source 
var GetSnapshotTS = func(txnOp client.TxnOperator) timestamp.Timestamp {
	return txnOp.SnapshotTS()
}
View Source 
var GetTableDef = func(
	ctx context.Context,
	txnOp client.TxnOperator,
	cnEngine engine.Engine,
	tblId uint64,
) (*plan.TableDef, error) {
	ctx, cancel := context.WithTimeoutCause(ctx, time.Minute*5, moerr.CauseGetTableDef)
	defer cancel()
	_, _, rel, err := cnEngine.GetRelationById(ctx, txnOp, tblId)
	if err != nil {
		return nil, err
	}

	return rel.CopyTableDef(ctx), nil
}
View Source 
var GetTableDetector = func(cnUUID string) *TableDetector {
	once.Do(func() {
		detector = &TableDetector{
			Mp:                   make(map[uint32]TblMap),
			Callbacks:            make(map[string]TableCallback),
			exec:                 getSqlExecutor(cnUUID),
			CallBackAccountId:    make(map[string]uint32),
			SubscribedAccountIds: make(map[uint32][]string),
			CallBackDbName:       make(map[string][]string),
			SubscribedDbNames:    make(map[string][]string),
			CallBackTableName:    make(map[string][]string),
			SubscribedTableNames: make(map[string][]string),
			cdcStateManager:      NewCDCStateManager(),
			cleanupPeriod:        DefaultWatermarkCleanupPeriod,
			cleanupWarn:          DefaultCleanupWarnThreshold,
			nowFn:                time.Now,
		}
		detector.scanTableFn = detector.scanTable
	})

	if detector != nil && detector.scanTableFn == nil {
		detector.scanTableFn = detector.scanTable
	}

	return detector
}
View Source 
var GetTxn = func(
	ctx context.Context,
	cnEngine engine.Engine,
	txnOp client.TxnOperator,
) error {
	return cnEngine.New(ctx, txnOp)
}
View Source 
var GetTxnOp = func(
	ctx context.Context,
	cnEngine engine.Engine,
	cnTxnClient client.TxnClient,
	info string,
) (client.TxnOperator, error) {
	nowTs := cnEngine.LatestLogtailAppliedTime()
	createByOpt := client.WithTxnCreateBy(
		0,
		"",
		info,
		0)
	return cnTxnClient.New(ctx, nowTs, createByOpt)
}
View Source 
var NewSinker = func(
	sinkUri UriInfo,
	accountId uint64,
	taskId string,
	dbTblInfo *DbTableInfo,
	watermarkUpdater *CDCWatermarkUpdater,
	tableDef *plan.TableDef,
	retryTimes int,
	retryDuration time.Duration,
	ar *ActiveRoutine,
	maxSqlLength uint64,
	sendSqlTimeout string,
) (Sinker, error) {

	if sinkUri.SinkTyp == CDCSinkType_Console {
		return NewConsoleSinker(dbTblInfo, watermarkUpdater), nil
	}

	return CreateMysqlSinker2(
		sinkUri,
		accountId,
		taskId,
		dbTblInfo,
		watermarkUpdater,
		tableDef,
		retryTimes,
		retryDuration,
		ar,
		maxSqlLength,
		sendSqlTimeout,
	)
}
View Source 
var NewTableChangeStream = func(
	cnTxnClient client.TxnClient,
	cnEngine engine.Engine,
	mp *mpool.MPool,
	packerPool *fileservice.Pool[*types.Packer],
	accountId uint64,
	taskId string,
	tableInfo *DbTableInfo,
	sinker Sinker,
	watermarkUpdater WatermarkUpdater,
	tableDef *plan.TableDef,
	initSnapshotSplitTxn bool,
	runningReaders *sync.Map,
	startTs, endTs types.TS,
	noFull bool,
	frequency time.Duration,
	options ...TableChangeStreamOption,
) *TableChangeStream {

	if frequency <= 0 {
		frequency = DefaultFrequency
	}

	opts := defaultTableChangeStreamOptions()
	for _, opt := range options {
		if opt != nil {
			opt(opts)
		}
	}
	opts.fillDefaults()

	watermarkKey := &WatermarkKey{
		AccountId: accountId,
		TaskId:    taskId,
		DBName:    tableInfo.SourceDbName,
		TableName: tableInfo.SourceTblName,
	}

	txnManager := NewTransactionManager(
		sinker,
		watermarkUpdater,
		accountId,
		taskId,
		tableInfo.SourceDbName,
		tableInfo.SourceTblName,
	)

	insTsColIdx := len(tableDef.Cols) - 1
	insCompositedPkColIdx := len(tableDef.Cols) - 2
	delTsColIdx := 1
	delCompositedPkColIdx := 0

	if len(tableDef.Pkey.Names) == 1 {
		insCompositedPkColIdx = int(tableDef.Name2ColIndex[tableDef.Pkey.Names[0]])
	}

	dataProcessor := NewDataProcessor(
		sinker,
		txnManager,
		mp,
		packerPool,
		insTsColIdx,
		insCompositedPkColIdx,
		delTsColIdx,
		delCompositedPkColIdx,
		initSnapshotSplitTxn,
		accountId,
		taskId,
		tableInfo.SourceDbName,
		tableInfo.SourceTblName,
	)

	progressTracker := NewProgressTracker(
		accountId,
		taskId,
		tableInfo.SourceDbName,
		tableInfo.SourceTblName,
	)

	if attachable, ok := sinker.(interface{ AttachProgressTracker(*ProgressTracker) }); ok {
		attachable.AttachProgressTracker(progressTracker)
	}

	stream := &TableChangeStream{
		txnManager:                txnManager,
		dataProcessor:             dataProcessor,
		cnTxnClient:               cnTxnClient,
		cnEngine:                  cnEngine,
		mp:                        mp,
		packerPool:                packerPool,
		accountId:                 accountId,
		taskId:                    taskId,
		tableInfo:                 tableInfo,
		tableDef:                  tableDef,
		sinker:                    sinker,
		watermarkUpdater:          watermarkUpdater,
		watermarkKey:              watermarkKey,
		tick:                      time.NewTicker(frequency),
		frequency:                 frequency,
		runningReaders:            runningReaders,
		runningReaderKey:          GenDbTblKey(tableInfo.SourceDbName, tableInfo.SourceTblName),
		initSnapshotSplitTxn:      initSnapshotSplitTxn,
		startTs:                   startTs,
		endTs:                     endTs,
		noFull:                    noFull,
		insTsColIdx:               insTsColIdx,
		insCompositedPkColIdx:     insCompositedPkColIdx,
		delTsColIdx:               delTsColIdx,
		delCompositedPkColIdx:     delCompositedPkColIdx,
		progressTracker:           progressTracker,
		watermarkStallThreshold:   opts.watermarkStallThreshold,
		noProgressWarningInterval: opts.noProgressWarningInterval,
		lastWatermarkAdvance:      time.Now(),

		maxRetryCount:      opts.maxRetryCount,
		retryBackoffBase:   opts.retryBackoffBase,
		retryBackoffMax:    opts.retryBackoffMax,
		retryBackoffFactor: opts.retryBackoffFactor,
	}

	tableLabel := progressTracker.tableKey()
	v2.CdcTableStuckGauge.WithLabelValues(tableLabel).Set(0)
	v2.CdcTableLastActivityTimestamp.WithLabelValues(tableLabel).Set(float64(time.Now().Unix()))

	stream.start.Add(1)
	return stream
}

NewTableChangeStream creates a new table change stream

View Source 
var OpenDbConn = func(user, password string, ip string, port int, timeout string) (db *sql.DB, err error) {
	logutil.Info("cdc.util.open_db_conn", zap.String("timeout", timeout))
	cfg, err := makeMysqlConfig(user, password, ip, port, timeout)
	if err != nil {
		return nil, err
	}
	for i := 0; i < 3; i++ {
		if db, err = tryConn(cfg); err == nil {

			return
		}
		v2.CdcMysqlConnErrorCounter.Inc()
		time.Sleep(time.Second)
	}
	logutil.Error("cdc.util.open_db_conn_failed", zap.Error(err))
	return
}

Functions

func AddSingleQuotesJoin 

func AddSingleQuotesJoin(s []string) string

AddSingleQuotesJoin [a, b, c] -> 'a','b','c'

func AesCFBDecode 

func AesCFBDecode(ctx context.Context, data string) (string, error)

func AesCFBEncode 

func AesCFBEncode(data []byte) (string, error)

func AesCFBEncodeWithKey 

func AesCFBEncodeWithKey(data []byte, aesKey []byte) (string, error)

func FormatErrorMetadata 

func FormatErrorMetadata(meta *ErrorMetadata) string

FormatErrorMetadata formats error metadata to string

func GenDbTblKey 

func GenDbTblKey(dbName, tblName string) string

func GetInitDataKeySql 

func GetInitDataKeySql(kek string) (_ string, err error)

func IsErrorExpired 

func IsErrorExpired(meta *ErrorMetadata) bool

IsErrorExpired checks if a non-retryable error has expired Deprecated: Error expiration mechanism removed. Non-retryable errors now permanently block until manually cleared via Resume.

func IsInsertClause 

func IsInsertClause(inputSql string) (ok bool)

func IsInsertOnDuplicateUpdateClause 

func IsInsertOnDuplicateUpdateClause(inputSql string) (ok bool)

func IsPauseOrCancelError 

func IsPauseOrCancelError(errMsg string) bool

IsPauseOrCancelError checks if the error is a control signal (pause/cancel)

func IsSelectClause 

func IsSelectClause(inputSql string) (ok bool)

func IsUpdateClause 

func IsUpdateClause(inputSql string) (ok bool)

func JsonDecode 

func JsonDecode(jbytes string, value any) error

JsonDecode decodes the json bytes to objects

func JsonEncode 

func JsonEncode(value any) (string, error)

JsonEncode encodes the object to json

func NewMockSQLExecutor 

func NewMockSQLExecutor() *mockSQLExecutor

func NewMysqlSinker2 

func NewMysqlSinker2(
	executor *Executor,
	accountId uint64,
	taskId string,
	dbTblInfo *DbTableInfo,
	watermarkUpdater *CDCWatermarkUpdater,
	builder *CDCStatementBuilder,
	ar *ActiveRoutine,
) *mysqlSinker2

NewMysqlSinker2 creates a new improved MySQL sinker

func ParseFrequencyToDuration 

func ParseFrequencyToDuration(freq string) time.Duration

ParseFrequencyToDuration parses a frequency string (e.g., "1h", "30m") to time.Duration

func ShouldRetry 

func ShouldRetry(meta *ErrorMetadata) bool

ShouldRetry determines if an error should allow retry Design:

  • nil metadata: allow retry (no error)
  • Retryable error: allow retry if count <= MaxRetryCount
  • Non-retryable error: permanently block (user must manually clear via Resume)

func SplitDbTblKey 

func SplitDbTblKey(dbTblKey string) (dbName, tblName string)

Types

type ActiveRoutine 

type ActiveRoutine struct {
	sync.Mutex
	Pause  chan struct{}
	Cancel chan struct{}
	// contains filtered or unexported fields
}

func NewCdcActiveRoutine 

func NewCdcActiveRoutine() *ActiveRoutine

func (*ActiveRoutine) CloseCancel 

func (ar *ActiveRoutine) CloseCancel()

func (*ActiveRoutine) ClosePause 

func (ar *ActiveRoutine) ClosePause()

type AtomicBatch 

type AtomicBatch struct {
	Mp      *mpool.MPool
	Batches []*batch.Batch
	Rows    *btree.BTreeG[AtomicBatchRow]
	// contains filtered or unexported fields
}

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) Allocated 

func (bat *AtomicBatch) Allocated() int

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) DuplicateRows 

func (bat *AtomicBatch) DuplicateRows() int

func (*AtomicBatch) GetRowIterator 

func (bat *AtomicBatch) GetRowIterator() RowIterator

func (*AtomicBatch) RowCount 

func (bat *AtomicBatch) RowCount() int

func (*AtomicBatch) TotalRows 

func (bat *AtomicBatch) TotalRows() int

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 CDCStateManager 

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

func NewCDCStateManager 

func NewCDCStateManager() *CDCStateManager

func (*CDCStateManager) AddActiveRunner 

func (s *CDCStateManager) AddActiveRunner(tblInfo *DbTableInfo)

func (*CDCStateManager) PrintActiveRunners 

func (s *CDCStateManager) PrintActiveRunners(slowThreshold time.Duration)

func (*CDCStateManager) RemoveActiveRunner 

func (s *CDCStateManager) RemoveActiveRunner(tblInfo *DbTableInfo)

func (*CDCStateManager) UpdateActiveRunner 

func (s *CDCStateManager) UpdateActiveRunner(tblInfo *DbTableInfo, fromTs, toTs types.TS, start bool)

type CDCStatementBuilder 

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

CDCStatementBuilder constructs SQL statements for CDC sink operations.

Design Principles: - Stateless: Each method is a pure function with no side effects - Reusable: Can be shared across multiple sinkers - Testable: Easy to unit test in isolation - Efficient: Minimizes allocations through buffer reuse

The builder handles: - INSERT statements from snapshot/tail data - DELETE statements with single or composite primary keys - SQL size limits (splits large statements into multiple) - All MatrixOne data types serialization

func NewCDCStatementBuilder 

func NewCDCStatementBuilder(
	dbName, tableName string,
	tableDef *plan.TableDef,
	maxSQLSize uint64,
	isMO bool,
) (*CDCStatementBuilder, error)

NewCDCStatementBuilder creates a new SQL statement builder for a specific table

func (*CDCStatementBuilder) BuildDeleteSQL 

func (b *CDCStatementBuilder) BuildDeleteSQL(
	ctx context.Context,
	atmBatch *AtomicBatch,
	fromTs, toTs types.TS,
) ([][]byte, error)

BuildDeleteSQL constructs DELETE SQL statements from atomic batches

Returns multiple SQL statements if the batch is too large. Supports two formats: 1. Single PK: DELETE FROM t WHERE pk IN ((val1),(val2),...) 2. Composite PK (MO): DELETE FROM t WHERE pk1=a1 AND pk2=a2 OR pk1=b1 AND pk2=b2 ... 3. Composite PK (MySQL): DELETE FROM t WHERE (pk1,pk2) IN ((a1,a2),(b1,b2),...)

func (*CDCStatementBuilder) BuildInsertSQL 

func (b *CDCStatementBuilder) BuildInsertSQL(
	ctx context.Context,
	bat *batch.Batch,
	fromTs, toTs types.TS,
) ([][]byte, error)

BuildInsertSQL constructs INSERT SQL statements from a batch

Returns multiple SQL statements if the batch is too large to fit in one statement. Each returned []byte has 5-byte header reserved for mysql driver.

Format: 

/* [fromTs, toTs) */ REPLACE INTO `db`.`table` VALUES (row1),(row2),...;

func (*CDCStatementBuilder) EstimateDeleteRowSize 

func (b *CDCStatementBuilder) EstimateDeleteRowSize() int

EstimateDeleteRowSize estimates the SQL size needed for one delete condition

func (*CDCStatementBuilder) EstimateInsertRowSize 

func (b *CDCStatementBuilder) EstimateInsertRowSize() int

EstimateInsertRowSize estimates the SQL size needed for one row Used to determine batch sizing

type CDCWatermarkUpdater 

type CDCWatermarkUpdater struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

CDCWatermarkUpdater manages watermarks for CDC tasks with eventual consistency.

Consistency Model: - Watermarks are allowed to LAG behind actual data progress (acceptable: causes duplicate processing) - Watermarks MUST NEVER ADVANCE ahead of persisted data (forbidden: would cause data loss) - Updates are buffered in memory and persisted asynchronously via batch operations

Three-Tier Cache Architecture: 1. cacheUncommitted: In-memory write buffer, updated immediately on UpdateWatermarkOnly() 2. cacheCommitting: Transition state during database persistence 3. cacheCommitted: Synchronized with database, represents durable watermarks

Update Flow: 

UpdateWatermarkOnly() -> cacheUncommitted (instant, always succeeds)
                      -> cacheCommitting (moved by CronJob every 3s)
                      -> cacheCommitted + DB (after batch UPDATE succeeds)

Crash Recovery: - If system crashes before CronJob persists, watermarks in cacheUncommitted are lost - Next run will read old watermark from DB and re-process data (duplicate processing is acceptable) - This ensures watermarks never advance beyond persisted data

func GetCDCWatermarkUpdater 

func GetCDCWatermarkUpdater(
	cnUUID string,
	executor ie.InternalExecutor,
) *CDCWatermarkUpdater

func InitCDCWatermarkUpdaterForTest 

func InitCDCWatermarkUpdaterForTest(t *testing.T) (*CDCWatermarkUpdater, *mockSQLExecutor)

func NewCDCWatermarkUpdater 

func NewCDCWatermarkUpdater(
	name string,
	ie ie.InternalExecutor,
	opts ...UpdateOption,
) *CDCWatermarkUpdater

func (*CDCWatermarkUpdater) ForceFlush 

func (u *CDCWatermarkUpdater) ForceFlush(ctx context.Context) (err error)

func (*CDCWatermarkUpdater) GetCommitFailureCount 

func (u *CDCWatermarkUpdater) GetCommitFailureCount(key *WatermarkKey) uint32

GetCommitFailureCount returns the number of consecutive commit failures for the given key.

func (*CDCWatermarkUpdater) GetFromCache 

func (u *CDCWatermarkUpdater) GetFromCache(
	ctx context.Context,
	key *WatermarkKey,
) (watermark types.TS, err error)

GetFromCache retrieves the latest watermark from the three-tier cache.

Lookup Priority (from newest to oldest): 1. cacheUncommitted - most recent updates, not yet persisted 2. cacheCommitting - updates being persisted to database 3. cacheCommitted - synchronized with database

Returns ErrNoWatermarkFound if the key doesn't exist in any cache tier. This can happen when: - A new CDC task is starting for the first time - CronJob failed and caches were cleared (watermarks lost, acceptable by design)

func (*CDCWatermarkUpdater) GetOrAddCommitted 

func (u *CDCWatermarkUpdater) GetOrAddCommitted(
	ctx context.Context,
	key *WatermarkKey,
	watermark *types.TS,
) (ret types.TS, err error)

GetOrAddCommitted retrieves the persisted watermark from database, or adds it if not exists.

Used for CDC task initialization to determine the starting watermark: - If watermark exists in database: Return the persisted value (resume from last position) - If watermark doesn't exist: Add the provided watermark to database (new task starting)

Fast Path: - Checks cacheCommitted first to avoid database query if watermark is already in memory - Returns immediately if cached watermark >= requested watermark

Slow Path (Cache Miss): - Enqueues a job to read watermark from database - If found: Updates cacheCommitted and returns persisted value - If not found: Inserts new watermark record and returns it

Concurrency: Assumes no concurrent writes to the same key (single reader per table)

func (*CDCWatermarkUpdater) IsCircuitBreakerOpen 

func (u *CDCWatermarkUpdater) IsCircuitBreakerOpen(key *WatermarkKey) bool

IsCircuitBreakerOpen returns true if the circuit breaker is currently open for the given key.

func (*CDCWatermarkUpdater) MarkTaskPaused 

func (u *CDCWatermarkUpdater) MarkTaskPaused(taskId string)

MarkTaskPaused marks a task as paused to block watermark updates This is called when a task is being paused to prevent race conditions

func (*CDCWatermarkUpdater) RemoveCachedWM 

func (u *CDCWatermarkUpdater) RemoveCachedWM(
	ctx context.Context,
	key *WatermarkKey,
) (err error)

func (*CDCWatermarkUpdater) Start 

func (u *CDCWatermarkUpdater) Start()

func (*CDCWatermarkUpdater) Stop 

func (u *CDCWatermarkUpdater) Stop()

func (*CDCWatermarkUpdater) UnmarkTaskPaused 

func (u *CDCWatermarkUpdater) UnmarkTaskPaused(taskId string)

UnmarkTaskPaused removes the pause mark from a task This is called when a task resumes or is cancelled

func (*CDCWatermarkUpdater) UpdateWatermarkErrMsg 

func (u *CDCWatermarkUpdater) UpdateWatermarkErrMsg(
	ctx context.Context,
	key *WatermarkKey,
	errMsg string,
	errorCtx *ErrorContext,
) (err error)

UpdateWatermarkErrMsg updates error message with automatic intelligent handling: - Control signal filtering (pause/cancel) - Retry count tracking and auto-increment - Auto-conversion to non-retryable after MaxRetryCount - Timestamp recording - Error expiration support

Parameters:

  • ctx: Context
  • key: Watermark key
  • errMsg: Error message (empty string to clear error)
  • errorCtx: Error context (can be nil for backward compatibility)

Call examples:

  • Retryable: UpdateWatermarkErrMsg(ctx, key, "table not found", &ErrorContext{IsRetryable: true})
  • Non-retryable: UpdateWatermarkErrMsg(ctx, key, "type mismatch", &ErrorContext{IsRetryable: false})
  • Clear: UpdateWatermarkErrMsg(ctx, key, "", nil)
  • Legacy: UpdateWatermarkErrMsg(ctx, key, "retryable error:xxx", nil) // Auto-parsed

Design: Uses in-memory cache to avoid synchronous SQL queries, preserving the lazy batch processing design of WatermarkUpdater

func (*CDCWatermarkUpdater) UpdateWatermarkOnly 

func (u *CDCWatermarkUpdater) UpdateWatermarkOnly(
	ctx context.Context,
	key *WatermarkKey,
	watermark *types.TS,
) (err error)

UpdateWatermarkOnly buffers a watermark update in memory without immediate persistence.

Consistency Guarantee: - This method is called ONLY AFTER data has been successfully committed to the database - It buffers the watermark in cacheUncommitted for later batch persistence - Always returns nil (never fails) to maintain the consistency model

Persistence Timing: - Watermark is persisted asynchronously by CronJob (default: every 3 seconds) - If system crashes before CronJob runs, the watermark update is lost - This is acceptable: next run will re-read from old watermark (duplicate processing is idempotent)

Why Always Return Nil: - By design, watermark lag is acceptable but advance is forbidden - Caller ensures data is committed BEFORE calling this method - Even if this buffer operation "fails" (system crash), watermark stays behind (safe) - Returning errors would complicate caller logic without improving consistency

type ChangeCollector 

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

ChangeCollector collects changes from engine Key responsibilities: 1. Wrap engine.ChangesHandle for cleaner interface 2. Provide typed change data (Snapshot/TailWip/TailDone/NoMoreData) 3. Handle resource cleanup

func NewChangeCollector 

func NewChangeCollector(
	changesHandle engine.ChangesHandle,
	mp *mpool.MPool,
	fromTs, toTs types.TS,
	accountId uint64,
	taskId string,
	dbName string,
	tableName string,
) *ChangeCollector

NewChangeCollector creates a new change collector

func (*ChangeCollector) Close 

func (cc *ChangeCollector) Close() error

Close closes the change collector

func (*ChangeCollector) GetFromTs 

func (cc *ChangeCollector) GetFromTs() types.TS

GetFromTs returns the fromTs of this collection

func (*ChangeCollector) GetToTs 

func (cc *ChangeCollector) GetToTs() types.TS

GetToTs returns the toTs of this collection

func (*ChangeCollector) IsClosed 

func (cc *ChangeCollector) IsClosed() bool

IsClosed returns true if the collector is closed

func (*ChangeCollector) Next 

func (cc *ChangeCollector) Next(ctx context.Context) (*ChangeData, error)

Next retrieves the next batch of changes Returns: - *ChangeData: The change data (nil if error or no more data) - error: Any error encountered

When both insert and delete batches are nil and no error: - Type will be ChangeTypeNoMoreData

type ChangeData 

type ChangeData struct {
	// Type of change
	Type ChangeType

	// Insert data (new rows)
	InsertBatch *batch.Batch

	// Delete data (tombstones)
	DeleteBatch *batch.Batch

	// Original engine hint
	Hint engine.ChangesHandle_Hint
}

ChangeData represents a batch of change data

func (*ChangeData) Clean 

func (cd *ChangeData) Clean(mp *mpool.MPool)

Clean cleans up the batch data

func (*ChangeData) HasData 

func (cd *ChangeData) HasData() bool

HasData returns true if there is insert or delete data

type ChangeReader 

type ChangeReader interface {
	// Run starts the change reader in blocking mode
	// It should be called in a goroutine
	Run(ctx context.Context, ar *ActiveRoutine)

	// Close closes the reader and releases resources
	Close()

	// Wait blocks until the reader goroutine completes
	// This is used for graceful shutdown and testing
	Wait()

	// GetTableInfo returns the source table information
	// This is used to identify the reader and check for table ID changes
	GetTableInfo() *DbTableInfo
}

ChangeReader represents a CDC change data capture reader It monitors table changes and streams them to a downstream sinker

type ChangeType 

type ChangeType int

ChangeType represents the type of change data 

const (
	// ChangeTypeSnapshot - Snapshot data (full table scan at fromTs)
	ChangeTypeSnapshot ChangeType = iota
	// ChangeTypeTailWip - Incremental data (work in progress, accumulating)
	ChangeTypeTailWip
	// ChangeTypeTailDone - Incremental data (done, ready to commit)
	ChangeTypeTailDone
	// ChangeTypeNoMoreData - No more data available
	ChangeTypeNoMoreData
)

func (ChangeType) String 

func (ct ChangeType) String() string

type Command 

type Command struct {
	// Type of the command
	Type CommandType

	// Data for insert operations (snapshot)
	InsertBatch *batch.Batch

	// Data for insert/delete operations (tail)
	InsertAtmBatch *AtomicBatch
	DeleteAtmBatch *AtomicBatch

	// Mp is the mpool used to allocate InsertBatch (snapshot).
	// Needed for proper cleanup via batch.Clean(mp).
	Mp *mpool.MPool

	// Metadata for the command
	Meta CommandMetadata
}

Command represents a command sent from producer to consumer

func NewBeginCommand 

func NewBeginCommand() *Command

NewBeginCommand creates a BEGIN transaction command

func NewCommitCommand 

func NewCommitCommand() *Command

NewCommitCommand creates a COMMIT transaction command

func NewDummyCommand 

func NewDummyCommand() *Command

NewDummyCommand creates a dummy command for synchronization

func NewFlushCommand 

func NewFlushCommand(noMoreData bool, fromTs, toTs types.TS) *Command

NewFlushCommand creates a command to flush buffered SQL

func NewInsertBatchCommand 

func NewInsertBatchCommand(bat *batch.Batch, mp *mpool.MPool, fromTs, toTs types.TS) *Command

NewInsertBatchCommand creates a command to insert a batch of rows

func NewInsertDeleteBatchCommand 

func NewInsertDeleteBatchCommand(insertBatch, deleteBatch *AtomicBatch, fromTs, toTs types.TS) *Command

NewInsertDeleteBatchCommand creates a command to insert and delete rows

func NewRollbackCommand 

func NewRollbackCommand() *Command

NewRollbackCommand creates a ROLLBACK transaction command

func (*Command) Close 

func (c *Command) Close()

Close releases batch resources held by the command. Must be called after the command is processed (or skipped) to avoid memory leaks.

func (*Command) String 

func (c *Command) String() string

String returns a string representation of the command (for debugging)

func (*Command) Validate 

func (c *Command) Validate() error

Validate checks if the command is valid

type CommandMetadata 

type CommandMetadata struct {
	// Watermark range for this data
	FromTs types.TS
	ToTs   types.TS

	// Whether this is the last batch (no more data)
	NoMoreData bool
}

CommandMetadata contains metadata for a command

type CommandType 

type CommandType int

CommandType represents the type of command sent from producer to consumer 

const (
	// CmdBegin starts a new transaction
	CmdBegin CommandType = iota
	// CmdCommit commits the current transaction
	CmdCommit
	// CmdRollback rolls back the current transaction
	CmdRollback
	// CmdInsertBatch inserts a batch of rows (snapshot data)
	CmdInsertBatch
	// CmdInsertDeleteBatch inserts and deletes rows (tail data)
	CmdInsertDeleteBatch
	// CmdFlush flushes any buffered SQL statements
	CmdFlush
	// CmdDummy is a synchronization point (no-op)
	CmdDummy
)

func (CommandType) String 

func (ct CommandType) String() string

String returns the string representation of CommandType

type DataProcessor 

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

DataProcessor processes change data and sends to sinker Key responsibilities: 1. Process different types of changes (Snapshot/TailWip/TailDone/NoMoreData) 2. Accumulate TailWip/TailDone data into AtomicBatch 3. Coordinate with TransactionManager (decide when to BEGIN) 4. Send data to Sinker 5. Handle resource cleanup

func NewDataProcessor 

func NewDataProcessor(
	sinker Sinker,
	txnManager *TransactionManager,
	mp *mpool.MPool,
	packerPool *fileservice.Pool[*types.Packer],
	insTsColIdx int,
	insCompositedPkColIdx int,
	delTsColIdx int,
	delCompositedPkColIdx int,
	initSnapshotSplitTxn bool,
	accountId uint64,
	taskId string,
	dbName string,
	tableName string,
) *DataProcessor

NewDataProcessor creates a new data processor

func (*DataProcessor) Cleanup 

func (dp *DataProcessor) Cleanup()

Cleanup cleans up any remaining resources This should be called in defer to ensure cleanup even on errors This method is safe to call concurrently and is idempotent

func (*DataProcessor) GetDeleteAtmBatch 

func (dp *DataProcessor) GetDeleteAtmBatch() *AtomicBatch

GetDeleteAtmBatch returns the current delete atomic batch (for testing)

func (*DataProcessor) GetInsertAtmBatch 

func (dp *DataProcessor) GetInsertAtmBatch() *AtomicBatch

GetInsertAtmBatch returns the current insert atomic batch (for testing)

func (*DataProcessor) ProcessChange 

func (dp *DataProcessor) ProcessChange(ctx context.Context, data *ChangeData) error

ProcessChange processes a single ChangeData Returns error if processing fails

func (*DataProcessor) SetTransactionRange 

func (dp *DataProcessor) SetTransactionRange(fromTs, toTs types.TS)

SetTransactionRange sets the from/to timestamps for the current transaction

type DbTableInfo 

type DbTableInfo struct {
	SourceDbId      uint64
	SourceDbName    string
	SourceTblId     uint64
	SourceTblName   string
	SourceCreateSql string

	SinkDbName  string
	SinkTblName string

	IdChanged bool
}

func (DbTableInfo) Clone 

func (info DbTableInfo) Clone() *DbTableInfo

func (DbTableInfo) OnlyDiffinTblId 

func (info DbTableInfo) OnlyDiffinTblId(t *DbTableInfo) bool

func (DbTableInfo) String 

func (info DbTableInfo) String() string

type DecoderOutput 

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

func (*DecoderOutput) Close 

func (d *DecoderOutput) Close()

Close releases batch resources held by the DecoderOutput. Called when ownership was NOT transferred to a Command (early-return paths in Sink).

type ErrorContext 

type ErrorContext struct {
	IsRetryable     bool // Whether the error is retryable
	IsPauseOrCancel bool // Whether this is a pause/cancel control signal (optional, auto-detected if not set)
}

ErrorContext provides context for error message updates

type ErrorMetadata 

type ErrorMetadata struct {
	IsRetryable bool      // Whether this error is retryable
	RetryCount  int       // Number of retry attempts
	FirstSeen   time.Time // When first seen
	LastSeen    time.Time // When last seen
	Message     string    // Error message
}

ErrorMetadata stores error metadata parsed from database

func BuildErrorMetadata 

func BuildErrorMetadata(old *ErrorMetadata, record *ErrorRecord) *ErrorMetadata

BuildErrorMetadata builds new metadata based on old metadata and new error record

func ParseErrorMetadata 

func ParseErrorMetadata(errMsg string) *ErrorMetadata

ParseErrorMetadata parses error message to metadata Format:

  • Retryable: "R:count:firstSeen:lastSeen:message"
  • Non-retryable: "N:firstSeen:message"
  • Legacy retryable: "retryable error:message"
  • Legacy non-retryable: "message"

type ErrorRecord 

type ErrorRecord struct {
	Error           error     // The original error
	IsRetryable     bool      // Whether this error is retryable
	IsPauseOrCancel bool      // Whether this is a pause/cancel control signal
	Timestamp       time.Time // When the error occurred
}

ErrorRecord represents a CDC error with metadata for recording

type Executor 

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

Executor manages database connection, transaction lifecycle, and SQL execution.

Key Features: - Nil-safe transaction operations (idempotent Begin/Commit/Rollback) - Automatic retry with exponential backoff - Clear transaction state management - No transaction leaks (always cleans up on commit/rollback)

Transaction Guarantees: - BeginTx() can only be called when no active transaction exists - CommitTx()/RollbackTx() are idempotent (safe to call even if tx is nil) - After Commit/Rollback, transaction is always set to nil

func NewExecutor 

func NewExecutor(
	user, password string,
	ip string,
	port int,
	retryTimes int,
	retryDuration time.Duration,
	timeout string,
	doRecord bool,
) (*Executor, error)

NewExecutor creates a new Executor with database connection

func (*Executor) BeginTx 

func (e *Executor) BeginTx(ctx context.Context) error

BeginTx starts a new transaction

Returns error if: - A transaction is already active - Database connection failed

func (*Executor) Close 

func (e *Executor) Close() error

Close closes the database connection and rolls back any active transaction

func (*Executor) CommitTx 

func (e *Executor) CommitTx(ctx context.Context) error

CommitTx commits the current transaction

Idempotent: Returns nil if no active transaction exists Always sets e.tx to nil after commit (successful or not)

func (*Executor) Connect 

func (e *Executor) Connect() error

Connect establishes a database connection

func (*Executor) ExecSQL 

func (e *Executor) ExecSQL(
	ctx context.Context,
	ar *ActiveRoutine,
	sqlBuf []byte,
	needRetry bool,
) error

ExecSQL executes a SQL statement

If a transaction is active, executes within the transaction. Otherwise, executes as a standalone statement.

Parameters: - sqlBuf: SQL statement with 5-byte header (for mysql driver reuse) - needRetry: Whether to retry on failure - ar: ActiveRoutine for cancellation support

func (*Executor) ExecSimpleSQL 

func (e *Executor) ExecSimpleSQL(ctx context.Context, sql string) error

ExecSimpleSQL executes a simple SQL statement outside of transaction Used for initialization (CREATE DATABASE, USE db, etc.)

func (*Executor) GetMaxAllowedPacket 

func (e *Executor) GetMaxAllowedPacket() (uint64, error)

GetMaxAllowedPacket queries the database for max_allowed_packet setting

func (*Executor) HasActiveTx 

func (e *Executor) HasActiveTx() bool

HasActiveTx returns true if there's an active transaction

func (*Executor) RollbackTx 

func (e *Executor) RollbackTx(ctx context.Context) error

RollbackTx rolls back the current transaction

Idempotent: Returns nil if no active transaction exists Always sets e.tx to nil after rollback (successful or not)

type InternalExecResultForTest 

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

func (*InternalExecResultForTest) Column 

func (res *InternalExecResultForTest) Column(ctx context.Context, i uint64) (name string, typ uint8, signed bool, err error)

func (*InternalExecResultForTest) ColumnCount 

func (res *InternalExecResultForTest) ColumnCount() uint64

func (*InternalExecResultForTest) Error 

func (res *InternalExecResultForTest) Error() error

func (*InternalExecResultForTest) GetFloat64 

func (res *InternalExecResultForTest) GetFloat64(ctx context.Context, ridx uint64, cid uint64) (float64, error)

func (*InternalExecResultForTest) GetString 

func (res *InternalExecResultForTest) GetString(ctx context.Context, i uint64, j uint64) (string, error)

func (*InternalExecResultForTest) GetUint64 

func (res *InternalExecResultForTest) GetUint64(ctx context.Context, i uint64, j uint64) (uint64, error)

func (*InternalExecResultForTest) Row 

func (res *InternalExecResultForTest) Row(ctx context.Context, i uint64) ([]interface{}, error)

func (*InternalExecResultForTest) RowCount 

func (res *InternalExecResultForTest) RowCount() uint64

func (*InternalExecResultForTest) Value 

func (res *InternalExecResultForTest) Value(ctx context.Context, ridx uint64, cidx uint64) (interface{}, error)

type MysqlResultSetForTest 

type MysqlResultSetForTest struct {
	//column information
	Columns []string
	//column name --> column index
	Name2Index map[string]uint64
	//data
	Data [][]interface{}
}

type ObservabilityManager 

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

ObservabilityManager manages all observability components for CDC

func GetObservabilityManager 

func GetObservabilityManager() *ObservabilityManager

GetObservabilityManager returns the global observability manager (singleton)

func (*ObservabilityManager) GetProgressMonitor 

func (om *ObservabilityManager) GetProgressMonitor() *ProgressMonitor

GetProgressMonitor returns the progress monitor

func (*ObservabilityManager) RegisterTableStream 

func (om *ObservabilityManager) RegisterTableStream(stream *TableChangeStream)

RegisterTableStream registers a table stream for monitoring

func (*ObservabilityManager) Start 

func (om *ObservabilityManager) Start()

Start starts the observability manager

func (*ObservabilityManager) Stop 

func (om *ObservabilityManager) Stop()

Stop stops the observability manager

func (*ObservabilityManager) UnregisterTableStream 

func (om *ObservabilityManager) UnregisterTableStream(stream *TableChangeStream)

UnregisterTableStream unregisters a table stream from monitoring

type OutputType 

type OutputType int
const (
	OutputTypeSnapshot OutputType = iota
	OutputTypeTail
)

func (OutputType) String 

func (t OutputType) String() string

type ParseResult 

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

func ParseInsert 

func ParseInsert(inputSql string) (result ParseResult, err error)

func ParseInsertOnDuplicateUpdate 

func ParseInsertOnDuplicateUpdate(inputSql string) (result ParseResult, err error)

func ParseSelectByPKs 

func ParseSelectByPKs(inputSql string) (result ParseResult, err error)

func ParseUpdate 

func ParseUpdate(inputSql string) (result ParseResult, err error)

only support update with where by pk

type PatternTable 

type PatternTable struct {
	Database string `json:"database"`
	Table    string `json:"table"`
}

func (PatternTable) String 

func (table PatternTable) String() string

type PatternTuple 

type PatternTuple struct {
	Source       PatternTable `json:"Source"`
	Sink         PatternTable `json:"Sink"`
	OriginString string       `json:"-"`
	Reserved     string       `json:"reserved"`
}

func (*PatternTuple) String 

func (tuple *PatternTuple) String() string

type PatternTuples 

type PatternTuples struct {
	Pts      []*PatternTuple `json:"pts"`
	Reserved string          `json:"reserved"`
}

func (*PatternTuples) Append 

func (pts *PatternTuples) Append(pt *PatternTuple)

func (*PatternTuples) String 

func (pts *PatternTuples) String() string

type ProgressMonitor 

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

ProgressMonitor monitors multiple progress trackers and detects stuck tables

func NewProgressMonitor 

func NewProgressMonitor(checkInterval, stuckThreshold, logInterval time.Duration) *ProgressMonitor

NewProgressMonitor creates a new progress monitor

func (*ProgressMonitor) Register 

func (pm *ProgressMonitor) Register(tracker *ProgressTracker)

Register registers a progress tracker

func (*ProgressMonitor) Start 

func (pm *ProgressMonitor) Start()

Start starts the monitoring goroutine

func (*ProgressMonitor) Stop 

func (pm *ProgressMonitor) Stop()

Stop stops the monitoring goroutine

func (*ProgressMonitor) Unregister 

func (pm *ProgressMonitor) Unregister(tracker *ProgressTracker)

Unregister unregisters a progress tracker

type ProgressTracker 

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

ProgressTracker tracks the progress of CDC processing for a single table This provides detailed observability into what's happening at each stage

func NewProgressTracker 

func NewProgressTracker(accountId uint64, taskId, dbName, tableName string) *ProgressTracker

NewProgressTracker creates a new progress tracker

func (*ProgressTracker) CheckStuck 

func (pt *ProgressTracker) CheckStuck(stuckThreshold time.Duration) (bool, string)

CheckStuck detects if the tracker is stuck (no progress for a long time)

func (*ProgressTracker) EndRound 

func (pt *ProgressTracker) EndRound(success bool, err error)

EndRound marks the end of a processing round

func (*ProgressTracker) GetState 

func (pt *ProgressTracker) GetState() (state string, duration time.Duration)

GetState returns the current state and how long it has been in this state

func (*ProgressTracker) GetStats 

func (pt *ProgressTracker) GetStats() map[string]interface{}

GetStats returns current statistics

func (*ProgressTracker) LogProgress 

func (pt *ProgressTracker) LogProgress()

LogProgress logs the current progress (forced)

func (*ProgressTracker) RecordBatch 

func (pt *ProgressTracker) RecordBatch(rows, bytes uint64)

RecordBatch records processing of a batch

func (*ProgressTracker) RecordError 

func (pt *ProgressTracker) RecordError(err error)

RecordError records an error

func (*ProgressTracker) RecordRetry 

func (pt *ProgressTracker) RecordRetry()

RecordRetry records a retry attempt

func (*ProgressTracker) RecordSQL 

func (pt *ProgressTracker) RecordSQL(count uint64)

RecordSQL records execution of SQL statements by the sinker

func (*ProgressTracker) RecordTransaction 

func (pt *ProgressTracker) RecordTransaction()

RecordTransaction records a committed transaction

func (*ProgressTracker) SetState 

func (pt *ProgressTracker) SetState(state string)

SetState updates the current state

func (*ProgressTracker) SetTargetWatermark 

func (pt *ProgressTracker) SetTargetWatermark(target types.TS)

SetTargetWatermark sets the target watermark

func (*ProgressTracker) StartRound 

func (pt *ProgressTracker) StartRound(fromTs, toTs types.TS)

StartRound marks the start of a new processing round

func (*ProgressTracker) TableKey 

func (pt *ProgressTracker) TableKey() string

TableKey returns the table key for metrics labeling (public method)

func (*ProgressTracker) UpdateWatermark 

func (pt *ProgressTracker) UpdateWatermark(newWatermark types.TS)

UpdateWatermark records a watermark update

type ReadContext 

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

ReadContext holds the context for a read operation

func NewReadContext 

func NewReadContext(ctx context.Context, txnOp client.TxnOperator, packer *types.Packer) *ReadContext

NewReadContext creates a new read context

func (*ReadContext) GetTracker 

func (rc *ReadContext) GetTracker() *TransactionTracker

GetTracker returns the transaction tracker

func (*ReadContext) SetTracker 

func (rc *ReadContext) SetTracker(tracker *TransactionTracker)

SetTracker sets the transaction tracker

type ReaderState 

type ReaderState int32

ReaderState represents the state of a table reader 

const (
	// ReaderStateIdle - Reader is idle, no active transaction
	ReaderStateIdle ReaderState = iota
	// ReaderStateReading - Reader is actively reading changes
	ReaderStateReading
	// ReaderStateProcessing - Reader is processing and sending data to sinker
	ReaderStateProcessing
	// ReaderStateCommitting - Reader is committing transaction
	ReaderStateCommitting
	// ReaderStateError - Reader encountered an error
	ReaderStateError
)

func (ReaderState) String 

func (s ReaderState) 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
)

type Sink 

type Sink interface {
	Send(ctx context.Context, ar *ActiveRoutine, sqlBuf []byte, needRetry bool) error
	SendBegin(ctx context.Context) error
	SendCommit(ctx context.Context) error
	SendRollback(ctx context.Context) error
	Reset()
	Close()
}

Sink represents the destination mysql or matrixone

type Sinker 

type Sinker interface {
	Run(ctx context.Context, ar *ActiveRoutine)
	Sink(ctx context.Context, data *DecoderOutput)
	SendBegin()
	SendCommit()
	SendRollback()
	// SendDummy to guarantee the last sql is sent
	SendDummy()
	// Error must be called after Sink
	Error() error
	ClearError()
	Reset()
	Close()
}

Sinker manages and drains the sql parts

func NewConsoleSinker 

func NewConsoleSinker(
	dbTblInfo *DbTableInfo,
	watermarkUpdater *CDCWatermarkUpdater,
) Sinker

type StateManager 

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

StateManager manages the reader state atomically

func NewStateManager 

func NewStateManager() *StateManager

NewStateManager creates a new state manager

func (*StateManager) CompareAndSwapState 

func (sm *StateManager) CompareAndSwapState(old, new ReaderState) bool

CompareAndSwapState atomically compares and swaps the state

func (*StateManager) GetState 

func (sm *StateManager) GetState() ReaderState

GetState returns the current reader state

func (*StateManager) SetState 

func (sm *StateManager) SetState(state ReaderState)

SetState sets the reader state

type TableCallback 

type TableCallback func(map[uint32]TblMap) error

type TableChangeStream 

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

TableChangeStream captures and streams table changes for CDC This is a complete CDC data pipeline that: - Monitors table changes continuously - Manages transactions with dual-layer safety - Processes and transforms change data - Handles retries and error recovery - Maintains watermarks and state

func (*TableChangeStream) Close 

func (s *TableChangeStream) Close()

Close closes the change stream

func (*TableChangeStream) GetRetryable 

func (s *TableChangeStream) GetRetryable() bool

GetRetryable returns the retryable flag in a thread-safe way

func (*TableChangeStream) GetTableInfo 

func (s *TableChangeStream) GetTableInfo() *DbTableInfo

GetTableInfo returns the source table information

func (*TableChangeStream) Run 

func (s *TableChangeStream) Run(ctx context.Context, ar *ActiveRoutine)

Run starts the change stream

func (*TableChangeStream) Wait 

func (s *TableChangeStream) Wait()

Info returns the table information Wait blocks until the reader goroutine completes

type TableChangeStreamOption 

type TableChangeStreamOption func(*tableChangeStreamOptions)

func WithMaxRetryCount 

func WithMaxRetryCount(count int) TableChangeStreamOption

WithMaxRetryCount configures the maximum retry count before converting retryable errors to non-retryable.

func WithNoProgressWarningInterval 

func WithNoProgressWarningInterval(interval time.Duration) TableChangeStreamOption

WithNoProgressWarningInterval configures how frequently warnings are emitted when snapshot timestamps stall.

func WithRetryBackoff 

func WithRetryBackoff(base, max time.Duration, factor float64) TableChangeStreamOption

WithRetryBackoff configures exponential backoff parameters for retryable errors.

func WithWatermarkStallThreshold 

func WithWatermarkStallThreshold(threshold time.Duration) TableChangeStreamOption

WithWatermarkStallThreshold configures how long snapshot stagnation is tolerated before surfacing an error.

type TableDetector 

type TableDetector struct {
	Mp        map[uint32]TblMap
	Callbacks map[string]TableCallback

	CallBackAccountId    map[string]uint32
	SubscribedAccountIds map[uint32][]string

	// taskname -> [db1, db2 ...]
	CallBackDbName map[string][]string
	// dbname -> [taska, taskb ...]
	SubscribedDbNames map[string][]string

	// taskname -> [tbl1, tbl2 ...]
	CallBackTableName map[string][]string
	// tablename -> [taska, taskb ...]
	SubscribedTableNames map[string][]string
	// contains filtered or unexported fields
}

func (*TableDetector) Close 

func (s *TableDetector) Close()

func (*TableDetector) IsTaskRegistered 

func (s *TableDetector) IsTaskRegistered(id string) bool

IsTaskRegistered checks if a task is already registered

func (*TableDetector) RegisterIfAbsent 

func (s *TableDetector) RegisterIfAbsent(id string, accountId uint32, dbs []string, tables []string, cb TableCallback) bool

RegisterIfAbsent registers the task only if it has not been registered before. Returns true when registration succeeds, false if the task already exists.

func (*TableDetector) UnRegister 

func (s *TableDetector) UnRegister(id string)

type TableDetectorOption 

type TableDetectorOption func(*TableDetectorOptions)

func WithTableDetectorCleanupPeriod 

func WithTableDetectorCleanupPeriod(d time.Duration) TableDetectorOption

func WithTableDetectorCleanupWarnThreshold 

func WithTableDetectorCleanupWarnThreshold(d time.Duration) TableDetectorOption

func WithTableDetectorPrintInterval 

func WithTableDetectorPrintInterval(d time.Duration) TableDetectorOption

func WithTableDetectorSlowThreshold 

func WithTableDetectorSlowThreshold(d time.Duration) TableDetectorOption

type TableDetectorOptions 

type TableDetectorOptions struct {
	SlowThreshold        time.Duration
	PrintInterval        time.Duration
	CleanupPeriod        time.Duration
	CleanupWarnThreshold time.Duration
}

type TableIterationState 

type TableIterationState struct {
	CreateAt time.Time
	EndAt    time.Time
	FromTs   types.TS
	ToTs     types.TS
}

type TaskId 

type TaskId = uuid.UUID

func NewTaskId 

func NewTaskId() TaskId

type TblMap 

type TblMap map[string]*DbTableInfo

TblMap key is dbName.tableName, e.g. db1.t1

type TransactionManager 

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

TransactionManager manages the transaction lifecycle Key responsibilities: 1. Track transaction state using TransactionTracker 2. Interact with Sinker (SendBegin/Commit/Rollback) 3. Interact with WatermarkUpdater (update watermark) 4. Implement dual-layer safety (tracker + watermark)

Concurrency & Locking:

  • All PUBLIC methods on TransactionManager are serialized by an internal mutex. This guarantees safe access and mutation of the internal TransactionTracker.
  • DO NOT call any other PUBLIC TransactionManager API while holding the mutex. If a public method needs to rollback while holding the lock, it MUST call the private rollbackLocked instead of the public RollbackTransaction to avoid re-entrant locking and potential deadlocks.

func NewTransactionManager 

func NewTransactionManager(
	sinker Sinker,
	watermarkUpdater WatermarkUpdater,
	accountId uint64,
	taskId string,
	dbName string,
	tableName string,
) *TransactionManager

NewTransactionManager creates a new transaction manager

func (*TransactionManager) BeginTransaction 

func (tm *TransactionManager) BeginTransaction(ctx context.Context, fromTs, toTs types.TS) error

BeginTransaction starts a new transaction This should be called when we have data to send

func (*TransactionManager) CommitTransaction 

func (tm *TransactionManager) CommitTransaction(ctx context.Context) error

CommitTransaction commits the current transaction Key steps (ORDER MATTERS): 1. Send COMMIT to sinker 2. Update watermark (persistent proof) 3. Mark tracker as committed (memory state)

func (*TransactionManager) EnsureCleanup 

func (tm *TransactionManager) EnsureCleanup(ctx context.Context) error

EnsureCleanup ensures proper transaction cleanup This implements the dual-layer safety check: Layer 1: Check tracker state (fast, explicit) Layer 2: Verify watermark (reliable, persistent)

func (*TransactionManager) GetTracker 

func (tm *TransactionManager) GetTracker() *TransactionTracker

GetTracker returns the current transaction tracker

func (*TransactionManager) Reset 

func (tm *TransactionManager) Reset()

Reset resets the transaction manager for a new transaction

func (*TransactionManager) RollbackTransaction 

func (tm *TransactionManager) RollbackTransaction(ctx context.Context) error

RollbackTransaction rolls back the current transaction

type TransactionTracker 

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

TransactionTracker tracks the lifecycle of a transaction This is a key improvement: explicit transaction state tracking instead of relying on implicit hasBegin flag

func NewTransactionTracker 

func NewTransactionTracker(fromTs, toTs types.TS) *TransactionTracker

NewTransactionTracker creates a new transaction tracker

func (*TransactionTracker) GetExpectedWatermark 

func (t *TransactionTracker) GetExpectedWatermark() types.TS

GetExpectedWatermark returns the expected watermark after commit

func (*TransactionTracker) GetFromTs 

func (t *TransactionTracker) GetFromTs() types.TS

GetFromTs returns the fromTs of this transaction

func (*TransactionTracker) GetToTs 

func (t *TransactionTracker) GetToTs() types.TS

GetToTs returns the toTs of this transaction

func (*TransactionTracker) IsCompleted 

func (t *TransactionTracker) IsCompleted() bool

IsCompleted returns true if transaction is fully completed (committed or rolled back)

func (*TransactionTracker) IsWatermarkUpdated 

func (t *TransactionTracker) IsWatermarkUpdated() bool

IsWatermarkUpdated returns whether watermark has been updated

func (*TransactionTracker) MarkBegin 

func (t *TransactionTracker) MarkBegin()

MarkBegin marks that BEGIN has been sent to sinker

func (*TransactionTracker) MarkCommit 

func (t *TransactionTracker) MarkCommit()

MarkCommit marks that COMMIT has been sent to sinker

func (*TransactionTracker) MarkRollback 

func (t *TransactionTracker) MarkRollback()

MarkRollback marks that ROLLBACK has been sent to sinker

func (*TransactionTracker) MarkWatermarkUpdated 

func (t *TransactionTracker) MarkWatermarkUpdated()

MarkWatermarkUpdated marks that watermark has been updated

func (*TransactionTracker) NeedsRollback 

func (t *TransactionTracker) NeedsRollback() bool

NeedsRollback returns true if transaction needs to be rolled back This is the key improvement: explicit check instead of relying on watermark

func (*TransactionTracker) Reset 

func (t *TransactionTracker) Reset(fromTs, toTs types.TS)

Reset resets the tracker for a new transaction

func (*TransactionTracker) UpdateToTs 

func (t *TransactionTracker) UpdateToTs(toTs types.TS)

UpdateToTs updates the transaction end timestamp This is used when multiple batches are processed in one transaction and we need to track the latest toTs

type UpdateOption 

type UpdateOption func(*CDCWatermarkUpdater)

func WithCronJobErrorSupressTimes 

func WithCronJobErrorSupressTimes(times uint64) UpdateOption

func WithCronJobInterval 

func WithCronJobInterval(interval time.Duration) UpdateOption

func WithCustomizedCronJob 

func WithCustomizedCronJob(fn func(ctx context.Context)) UpdateOption

func WithCustomizedScheduleJob 

func WithCustomizedScheduleJob(fn func(job *UpdaterJob) (err error)) UpdateOption

func WithExportStatsInterval 

func WithExportStatsInterval(interval time.Duration) UpdateOption

type UpdaterJob 

type UpdaterJob struct {
	tasks.Job
	Key       *WatermarkKey
	Watermark types.TS
	ErrMsg    string
}

func NewCommittingWMJob 

func NewCommittingWMJob(
	ctx context.Context,
) *UpdaterJob

func NewGetOrAddCommittedWMJob 

func NewGetOrAddCommittedWMJob(
	ctx context.Context,
	key *WatermarkKey,
	watermark *types.TS,
) *UpdaterJob

func NewRemoveCachedWMJob 

func NewRemoveCachedWMJob(
	ctx context.Context,
	key *WatermarkKey,
) *UpdaterJob

func NewUpdateWMErrMsgJob 

func NewUpdateWMErrMsgJob(
	ctx context.Context,
	key *WatermarkKey,
	errMsg string,
) *UpdaterJob

type UriInfo 

type UriInfo struct {
	SinkTyp       string `json:"_"`
	User          string `json:"user"`
	Password      string `json:"-"`
	Ip            string `json:"ip"`
	Port          int    `json:"port"`
	PasswordStart int    `json:"-"`
	PasswordEnd   int    `json:"-"`
	Reserved      string `json:"reserved"`
}

func ExtractUriInfo 

func ExtractUriInfo(
	ctx context.Context,
	uri string,
	uriPrefix string,
) (string, UriInfo, error)

ExtractUriInfo extracts the uriInfo return serialized uriInfo

func (*UriInfo) GetEncodedPassword 

func (info *UriInfo) GetEncodedPassword() (string, error)

func (*UriInfo) String 

func (info *UriInfo) String() string

type WatermarkKey 

type WatermarkKey struct {
	AccountId uint64
	TaskId    string
	DBName    string
	TableName string
}

func (*WatermarkKey) String 

func (k *WatermarkKey) String() string

type WatermarkResult 

type WatermarkResult struct {
	Watermark types.TS
	Ok        bool
}

type WatermarkUpdater 

type WatermarkUpdater interface {
	RemoveCachedWM(ctx context.Context, key *WatermarkKey) (err error)
	UpdateWatermarkErrMsg(ctx context.Context, key *WatermarkKey, errMsg string, errorCtx *ErrorContext) (err error)
	GetFromCache(ctx context.Context, key *WatermarkKey) (watermark types.TS, err error)
	GetOrAddCommitted(ctx context.Context, key *WatermarkKey, watermark *types.TS) (ret types.TS, err error)
	UpdateWatermarkOnly(ctx context.Context, key *WatermarkKey, watermark *types.TS) (err error)
	IsCircuitBreakerOpen(key *WatermarkKey) bool
	GetCommitFailureCount(key *WatermarkKey) uint32
}

WatermarkUpdater manages CDC watermarks

Source Files

View all Source files

Directories

Path Synopsis

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.