// Copyright 2024 PingCAP, Inc.
//
// 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,
// See the License for the specific language governing permissions and
// limitations under the License.

package mysql

import (
	"database/sql"
	"fmt"
	"strconv"
	"strings"

	"github.com/pingcap/log"
	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
	"github.com/pingcap/ticdc/pkg/config"
	"github.com/pingcap/ticdc/pkg/errors"
	"github.com/pingcap/ticdc/pkg/filter"
	"go.uber.org/zap"
)

const (
	// syncPointMetaTableID is a reserved table_id in ddl_ts_v1 used to record syncpoint state.
	// It is negative to avoid conflicting with any real table_id (including DDLSpanTableID=0).
	syncPointMetaTableID int64 = -1
)

// FlushDDLTsPre is used to flush ddl ts before the ddl event is sent to downstream.
// It's used to fix the potential data loss problem leading by the ddl ts event and ddl event can't be atomicly send to downstream.
//
// For example,
// If we don't flush ddl ts pre before the ddl event,
// it may happens that the ddl event is sent to downstream, and the server is down
// so the ddl ts event is not sent to downstream.
// Then when the server is up, we search the ddl ts table and find the ddl ts is not exist,
// and we may think the ddl event is not sent to downstream, so we will use the startTs with the last ddl ts.
// It will cause we use the wrong startTs and cause the data loss.
//
// Thus, we try to flush ddl ts pre first before the ddl event is sent to downstream,
// and after send the ddl ts, we update the ddl ts item to finished.
// It can maximum guarantee we use the correct startTs.
func (w *Writer) FlushDDLTsPre(event commonEvent.BlockEvent) error {
	if w.cfg.DryRun || !w.cfg.EnableDDLTs {
		return nil
	}
	err := w.createDDLTsTableIfNotExist()
	if err != nil {
		return err
	}
	return w.SendDDLTsPre(event)
}

func (w *Writer) FlushDDLTs(event commonEvent.BlockEvent) error {
	if w.cfg.DryRun || !w.cfg.EnableDDLTs {
		return nil
	}
	err := w.createDDLTsTableIfNotExist()
	if err != nil {
		return err
	}
	return w.SendDDLTs(event)
}

func (w *Writer) SendDDLTsPre(event commonEvent.BlockEvent) error {
	tx, err := w.db.BeginTx(w.ctx, nil)
	if err != nil {
		return errors.WrapError(errors.ErrMySQLTxnError, errors.WithMessage(err, "ddl ts table: begin Tx fail;"))
	}

	changefeedID := w.ChangefeedID.String()
	ticdcClusterID := config.GetGlobalServerConfig().ClusterID
	ddlTs := strconv.FormatUint(event.GetCommitTs(), 10)
	var tableIds []int64
	var isSyncpoint string

	// Syncpoint should not update ddl_ts for all tables. That causes O(table_count)
	// upserts (twice per syncpoint: pre and post) and becomes a bottleneck when the
	// number of tables is large. Instead, we only update a single reserved row to
	// represent the latest syncpoint state for this changefeed.
	if event.GetType() == commonEvent.TypeSyncPointEvent {
		tableIds = []int64{syncPointMetaTableID}
		isSyncpoint = "1"
	} else {
		isSyncpoint = "0"
		relatedTables := event.GetBlockedTables()

		switch relatedTables.InfluenceType {
		case commonEvent.InfluenceTypeNormal:
			tableIds = append(tableIds, relatedTables.TableIDs...)
		case commonEvent.InfluenceTypeDB:
			ids := w.tableSchemaStore.GetTableIdsByDB(relatedTables.SchemaID)
			tableIds = append(tableIds, ids...)
		case commonEvent.InfluenceTypeAll:
			ids := w.tableSchemaStore.GetAllTableIds()
			tableIds = append(tableIds, ids...)
		}

		addTables := event.GetNeedAddedTables()
		for _, table := range addTables {
			tableIds = append(tableIds, table.TableID)
		}
	}

	if len(tableIds) > 0 {
		err = w.execInsertItemBatches(tx, tableIds, ticdcClusterID, changefeedID, ddlTs, "0", isSyncpoint)
		if err != nil {
			log.Error("failed to write ddl ts table", zap.Error(err))
			err2 := tx.Rollback()
			if err2 != nil {
				log.Error("failed to write ddl ts table", zap.Error(err2))
			}
			return errors.WrapError(errors.ErrMySQLTxnError, errors.WithMessage(err, "failed to write ddl ts table; Exec Failed"))
		}
	} else {
		log.Error("table ids is empty when write ddl ts table, FIX IT", zap.Any("event", event))
	}

	err = tx.Commit()
	return errors.WrapError(errors.ErrMySQLTxnError, errors.WithMessage(err, "failed to write ddl ts table; Commit Fail;"))
}

func (w *Writer) SendDDLTs(event commonEvent.BlockEvent) error {
	tx, err := w.db.BeginTx(w.ctx, nil)
	if err != nil {
		return errors.WrapError(errors.ErrMySQLTxnError, errors.WithMessage(err, "ddl ts table: begin Tx fail;"))
	}

	changefeedID := w.ChangefeedID.String()
	ticdcClusterID := config.GetGlobalServerConfig().ClusterID

	ddlTs := strconv.FormatUint(event.GetCommitTs(), 10)
	var tableIds []int64
	var dropTableIds []int64
	var isSyncpoint string

	// See SendDDLTsPre: syncpoint only updates a single reserved row.
	if event.GetType() == commonEvent.TypeSyncPointEvent {
		tableIds = []int64{syncPointMetaTableID}
		isSyncpoint = "1"
	} else {
		isSyncpoint = "0"
		relatedTables := event.GetBlockedTables()

		switch relatedTables.InfluenceType {
		case commonEvent.InfluenceTypeNormal:
			tableIds = append(tableIds, relatedTables.TableIDs...)
		case commonEvent.InfluenceTypeDB:
			ids := w.tableSchemaStore.GetTableIdsByDB(relatedTables.SchemaID)
			tableIds = append(tableIds, ids...)
		case commonEvent.InfluenceTypeAll:
			ids := w.tableSchemaStore.GetAllTableIds()
			tableIds = append(tableIds, ids...)
		}

		dropTables := event.GetNeedDroppedTables()
		if dropTables != nil {
			switch dropTables.InfluenceType {
			case commonEvent.InfluenceTypeNormal:
				dropTableIds = append(dropTableIds, dropTables.TableIDs...)
			case commonEvent.InfluenceTypeDB:
				// for drop table, we will never delete the item of table trigger, so we get normal table ids for the schemaID.
				ids := w.tableSchemaStore.GetNormalTableIdsByDB(dropTables.SchemaID)
				dropTableIds = append(dropTableIds, ids...)
			case commonEvent.InfluenceTypeAll:
				// for drop table, we will never delete the item of table trigger, so we get normal table ids for the schemaID.
				ids := w.tableSchemaStore.GetAllNormalTableIds()
				dropTableIds = append(dropTableIds, ids...)
			}
		}

		addTables := event.GetNeedAddedTables()
		for _, table := range addTables {
			tableIds = append(tableIds, table.TableID)
		}
	}

	if len(tableIds) > 0 {
		err = w.execInsertItemBatches(tx, tableIds, ticdcClusterID, changefeedID, ddlTs, "1", isSyncpoint)
		if err != nil {
			log.Error("failed to write ddl ts table", zap.Error(err))
			err2 := tx.Rollback()
			if err2 != nil {
				log.Error("failed to write ddl ts table", zap.Error(err2))
			}
			return errors.WrapError(errors.ErrMySQLTxnError, errors.WithMessage(err, "failed to write ddl ts table; Exec Failed"))
		}
	} else {
		log.Error("table ids is empty when write ddl ts table, FIX IT", zap.Any("event", event))
	}

	if len(dropTableIds) > 0 {
		queries := dropItemQueries(dropTableIds, ticdcClusterID, changefeedID, w.maxDDLTsBatch)
		for _, query := range queries {
			log.Debug("send ddl ts table query", zap.String("query", query))

			if _, err = tx.Exec(query); err != nil {
				log.Error("failed to delete ddl ts item ", zap.Error(err))
				err2 := tx.Rollback()
				if err2 != nil {
					log.Error("failed to delete ddl ts item", zap.Error(err2))
				}
				return errors.WrapError(errors.ErrMySQLTxnError, errors.WithMessage(err, fmt.Sprintf("failed to delete ddl ts item; Query is %s", query)))
			}
		}
	}

	err = tx.Commit()
	return errors.WrapError(errors.ErrMySQLTxnError, errors.WithMessage(err, "failed to write ddl ts table; Commit Fail;"))
}

func (w *Writer) execInsertItemBatches(
	tx *sql.Tx,
	tableIds []int64,
	ticdcClusterID, changefeedID, ddlTs, finished, isSyncpoint string,
) error {
	if len(tableIds) == 0 {
		return nil
	}
	batchSize := w.maxDDLTsBatch
	for start := 0; start < len(tableIds); start += batchSize {
		end := start + batchSize
		if end > len(tableIds) {
			end = len(tableIds)
		}
		query := buildInsertItemQuery(tableIds[start:end], ticdcClusterID, changefeedID, ddlTs, finished, isSyncpoint)
		log.Debug("send ddl ts table query", zap.String("query", query))
		if _, err := tx.Exec(query); err != nil {
			return errors.WithMessage(err, fmt.Sprintf("failed to execute ddl ts insert chunk [%d:%d)", start, end))
		}
	}
	return nil
}

func buildInsertItemQuery(tableIds []int64, ticdcClusterID, changefeedID, ddlTs, finished, isSyncpoint string) string {
	var builder strings.Builder
	builder.WriteString("INSERT INTO ")
	builder.WriteString(filter.TiCDCSystemSchema)
	builder.WriteString(".")
	builder.WriteString(filter.DDLTsTable)
	builder.WriteString(" (ticdc_cluster_id, changefeed, ddl_ts, table_id, finished, is_syncpoint) VALUES ")
	for idx, tableId := range tableIds {
		if idx > 0 {
			builder.WriteString(", ")
		}
		builder.WriteString("('")
		builder.WriteString(ticdcClusterID)
		builder.WriteString("', '")
		builder.WriteString(changefeedID)
		builder.WriteString("', '")
		builder.WriteString(ddlTs)
		builder.WriteString("', ")
		builder.WriteString(strconv.FormatInt(tableId, 10))
		builder.WriteString(", ")
		builder.WriteString(finished)
		builder.WriteString(", ")
		builder.WriteString(isSyncpoint)
		builder.WriteString(")")
	}
	builder.WriteString(" ON DUPLICATE KEY UPDATE finished=VALUES(finished), ddl_ts=VALUES(ddl_ts), is_syncpoint=VALUES(is_syncpoint);")
	return builder.String()
}

func dropItemQueries(dropTableIds []int64, ticdcClusterID string, changefeedID string, maxBatch int) []string {
	result := make([]string, 0, (len(dropTableIds)+maxBatch-1)/maxBatch)
	for start := 0; start < len(dropTableIds); start += maxBatch {
		end := start + maxBatch
		if end > len(dropTableIds) {
			end = len(dropTableIds)
		}
		var builder strings.Builder
		builder.WriteString("DELETE FROM ")
		builder.WriteString(filter.TiCDCSystemSchema)
		builder.WriteString(".")
		builder.WriteString(filter.DDLTsTable)
		builder.WriteString(" WHERE (ticdc_cluster_id, changefeed, table_id) IN (")

		for idx, tableId := range dropTableIds[start:end] {
			if idx > 0 {
				builder.WriteString(", ")
			}
			builder.WriteString("('")
			builder.WriteString(ticdcClusterID)
			builder.WriteString("', '")
			builder.WriteString(changefeedID)
			builder.WriteString("', ")
			builder.WriteString(strconv.FormatInt(tableId, 10))
			builder.WriteString(")")
		}

		builder.WriteString(")")
		result = append(result, builder.String())
	}

	return result
}

// GetTableRecoveryInfo queries ddl_ts_v1 to determine recovery information for the given tables.
//
// It reads:
//  1. Per-table row: (table_id = tableID)
//  2. Syncpoint meta row: (table_id = syncPointMetaTableID)
//
// The syncpoint meta row is only applied when the per-table row exists. If the per-table
// row does not exist, we return 0 and let the caller use the input startTs directly.
//
// When both rows exist, we choose the row with larger ddl_ts (if equal, prefer the
// syncpoint meta row), and decode it as following rules:
//
//  1. finished=1: DDL and optional syncpoint completed normally
//     - Returns ddlTs
//     - skipSyncpointAtStartTs = is_syncpoint (skip if it was a syncpoint)
//     - skipDMLAsStartTs = false
//  2. finished=0, is_syncpoint=false: DDL not finished (crash during DDL)
//     - Returns ddlTs-1 to replay DDL at ddlTs
//     - skipDMLAsStartTs = true (skip already-written DML at ddlTs)
//  3. finished=0, is_syncpoint=true: Syncpoint not finished
//     - If there was a DDL at the same ts, it has already been executed
//     (because we only write syncpoint pre after DDL is fully completed)
//     - Returns ddlTs (no need to replay DDL even if it existed)
//     - skipSyncpointAtStartTs = false (replay syncpoint)
//     - skipDMLAsStartTs = false (process DML normally)
//
// Returns:
//   - startTsList: The startTs to use for each table
//   - skipSyncpointAtStartTsList: Whether to skip syncpoint events at startTs for each table
//   - skipDMLAsStartTsList: Whether to skip DML events at startTs+1 for each table
func (w *Writer) GetTableRecoveryInfo(tableIDs []int64) ([]int64, []bool, []bool, error) {
	retStartTsList := make([]int64, len(tableIDs))
	// when split table enabled, there may have some same tableID in tableIDs
	tableIdIdxMap := make(map[int64][]int, len(tableIDs))
	skipSyncpointAtStartTs := make([]bool, len(tableIDs))
	skipDMLAsStartTsList := make([]bool, len(tableIDs))
	for i, tableID := range tableIDs {
		tableIdIdxMap[tableID] = append(tableIdIdxMap[tableID], i)
		skipSyncpointAtStartTs[i] = false
		skipDMLAsStartTsList[i] = false
	}

	changefeedID := w.ChangefeedID.String()
	ticdcClusterID := config.GetGlobalServerConfig().ClusterID

	query := selectDDLTsQuery(tableIDs, ticdcClusterID, changefeedID)
	log.Info("query ddl ts table", zap.String("query", query))
	rows, err := w.db.Query(query)
	if err != nil {
		if errors.IsTableNotExistsErr(err) {
			// If this table is not existed, this means the table is first being synced
			log.Info("ddl ts table is not found",
				zap.String("keyspace", w.ChangefeedID.Keyspace()),
				zap.String("changefeedID", w.ChangefeedID.Name()),
				zap.Error(err))
			return retStartTsList, skipSyncpointAtStartTs, skipDMLAsStartTsList, nil
		}
		return retStartTsList, skipSyncpointAtStartTs, skipDMLAsStartTsList, errors.WrapError(errors.ErrMySQLTxnError, errors.WithMessage(err, fmt.Sprintf("failed to check ddl ts table; Query is %s", query)))
	}

	defer rows.Close()

	type ddlTsRow struct {
		ddlTs       int64
		finished    bool
		isSyncpoint bool
	}

	var (
		syncpointRow      ddlTsRow
		hasSyncpointRow   bool
		tableIDToDDLTsRow = make(map[int64]ddlTsRow, len(tableIDs))
	)

	for rows.Next() {
		var (
			ddlTs, tableId     int64
			finished, isSynced bool
		)
		err = rows.Scan(&tableId, &ddlTs, &finished, &isSynced)
		if err != nil {
			return retStartTsList, skipSyncpointAtStartTs, skipDMLAsStartTsList, errors.WrapError(errors.ErrMySQLTxnError, errors.WithMessage(err, fmt.Sprintf("failed to check ddl ts table; Query is %s", query)))
		}
		row := ddlTsRow{
			ddlTs:       ddlTs,
			finished:    finished,
			isSyncpoint: isSynced,
		}
		if tableId == syncPointMetaTableID {
			syncpointRow = row
			hasSyncpointRow = true
			continue
		}

		if _, ok := tableIdIdxMap[tableId]; !ok {
			continue
		}
		tableIDToDDLTsRow[tableId] = row
	}

	decodeRow := func(row ddlTsRow) (startTs int64, skipSyncpoint bool, skipDMLAsStartTs bool) {
		if row.finished {
			return row.ddlTs, row.isSyncpoint, false
		}
		if row.isSyncpoint {
			// Syncpoint not finished, replay it at the same ts.
			return row.ddlTs, false, false
		}
		// DDL not finished, replay it at ddlTs, but skip the already-written DML at ddlTs.
		return row.ddlTs - 1, false, true
	}

	for tableID, idxList := range tableIdIdxMap {
		tableRow, ok := tableIDToDDLTsRow[tableID]
		if !ok {
			continue
		}

		chosen := tableRow
		if hasSyncpointRow {
			// When ddl_ts is equal, prefer the syncpoint meta row. This keeps the
			// behavior consistent with the legacy per-table updates where the
			// syncpoint write overwrote the per-table row at the same ts.
			if syncpointRow.ddlTs >= tableRow.ddlTs {
				chosen = syncpointRow
			}
		}

		startTs, skipSyncpoint, skipDMLAsStartTs := decodeRow(chosen)
		for _, idx := range idxList {
			retStartTsList[idx] = startTs
			skipSyncpointAtStartTs[idx] = skipSyncpoint
			skipDMLAsStartTsList[idx] = skipDMLAsStartTs
		}
	}

	return retStartTsList, skipSyncpointAtStartTs, skipDMLAsStartTsList, nil
}

func selectDDLTsQuery(tableIDs []int64, ticdcClusterID string, changefeedID string) string {
	var builder strings.Builder
	builder.WriteString("SELECT table_id, ddl_ts, finished, is_syncpoint FROM ")
	builder.WriteString(filter.TiCDCSystemSchema)
	builder.WriteString(".")
	builder.WriteString(filter.DDLTsTable)
	builder.WriteString(" WHERE (ticdc_cluster_id, changefeed, table_id) IN (")

	for idx, tableID := range tableIDs {
		builder.WriteString("('")
		builder.WriteString(ticdcClusterID)
		builder.WriteString("', '")
		builder.WriteString(changefeedID)
		builder.WriteString("', ")
		builder.WriteString(strconv.FormatInt(tableID, 10))
		builder.WriteString(")")
		if idx < len(tableIDs)-1 {
			builder.WriteString(", ")
		}
	}
	if len(tableIDs) > 0 {
		builder.WriteString(", ")
	}
	// Always query the syncpoint meta row. It may or may not exist.
	builder.WriteString("('")
	builder.WriteString(ticdcClusterID)
	builder.WriteString("', '")
	builder.WriteString(changefeedID)
	builder.WriteString("', ")
	builder.WriteString(strconv.FormatInt(syncPointMetaTableID, 10))
	builder.WriteString(")")
	builder.WriteString(")")
	return builder.String()
}

func (w *Writer) RemoveDDLTsItem() error {
	tx, err := w.db.BeginTx(w.ctx, nil)
	if err != nil {
		return errors.WrapError(errors.ErrMySQLTxnError, errors.WithMessage(err, "select ddl ts table: begin Tx fail;"))
	}

	changefeedID := w.ChangefeedID.String()
	ticdcClusterID := config.GetGlobalServerConfig().ClusterID

	var builder strings.Builder
	builder.WriteString("DELETE FROM ")
	builder.WriteString(filter.TiCDCSystemSchema)
	builder.WriteString(".")
	builder.WriteString(filter.DDLTsTable)
	builder.WriteString(" WHERE (ticdc_cluster_id, changefeed) IN (")

	builder.WriteString("('")
	builder.WriteString(ticdcClusterID)
	builder.WriteString("', '")
	builder.WriteString(changefeedID)
	builder.WriteString("')")
	builder.WriteString(")")
	query := builder.String()

	_, err = tx.Exec(query)
	if err != nil {
		if errors.IsTableNotExistsErr(err) {
			if rbErr := tx.Rollback(); rbErr != nil {
				log.Error("failed to rollback", zap.Error(rbErr))
			}
			// If this table is not existed, this means the changefeed has not table, so we just return nil.
			log.Info("ddl ts table is not found when RemoveDDLTsItem",
				zap.String("keyspace", w.ChangefeedID.Keyspace()),
				zap.String("changefeedID", w.ChangefeedID.Name()),
				zap.Error(err))
			return nil
		}
		log.Error("failed to delete ddl ts item ", zap.Error(err))
		err2 := tx.Rollback()
		if err2 != nil {
			log.Error("failed to delete ddl ts item", zap.Error(err2))
		}
		return errors.WrapError(errors.ErrMySQLTxnError, errors.WithMessage(err, fmt.Sprintf("failed to delete ddl ts item; Query is %s", query)))
	}

	err = tx.Commit()
	return errors.WrapError(errors.ErrMySQLTxnError, errors.WithMessage(err, fmt.Sprintf("failed to delete ddl ts item; Query is %s", query)))
}

func (w *Writer) isDDLExecuted(tableID int64, ddlTs uint64) (bool, error) {
	changefeedID := w.ChangefeedID.String()
	ticdcClusterID := config.GetGlobalServerConfig().ClusterID

	// select * from xx where (ticdc_cluster_id, changefeed, table_id, ddl_ts) in (("xx","xx",x,x));
	var builder strings.Builder
	builder.WriteString("SELECT * FROM ")
	builder.WriteString(filter.TiCDCSystemSchema)
	builder.WriteString(".")
	builder.WriteString(filter.DDLTsTable)
	builder.WriteString(" WHERE (ticdc_cluster_id, changefeed, table_id, ddl_ts, finished) IN (")

	builder.WriteString("('")
	builder.WriteString(ticdcClusterID)
	builder.WriteString("', '")
	builder.WriteString(changefeedID)
	builder.WriteString("', ")
	builder.WriteString(strconv.FormatInt(tableID, 10))
	builder.WriteString(", ")
	builder.WriteString(strconv.FormatUint(ddlTs, 10))
	builder.WriteString(", ")
	builder.WriteString("1")
	builder.WriteString(")")
	builder.WriteString(")")
	query := builder.String()

	rows, err := w.db.Query(query)
	if err != nil {
		return false, errors.WrapError(errors.ErrMySQLTxnError, errors.WithMessage(err, fmt.Sprintf("failed to check ddl ts table; Query is %s", query)))
	}

	defer rows.Close()
	if rows.Next() {
		return true, nil
	}
	return false, nil
}

func (w *Writer) createTable(dbName string, tableName string, createTableQuery string) error {
	tx, err := w.db.BeginTx(w.ctx, nil)
	if err != nil {
		return errors.WrapError(errors.ErrMySQLTxnError, errors.WithMessage(err, fmt.Sprintf("create %s table: begin Tx fail;", tableName)))
	}

	_, err = tx.Exec("CREATE DATABASE IF NOT EXISTS " + dbName)
	if err != nil {
		errRollback := tx.Rollback()
		if errRollback != nil {
			log.Error("failed to rollback", zap.Any("tableName", tableName), zap.Error(errRollback))
		}
		return errors.WrapError(errors.ErrMySQLTxnError, errors.WithMessage(err, fmt.Sprintf("failed to create %s table;", tableName)))
	}
	_, err = tx.Exec("USE " + dbName)
	if err != nil {
		errRollback := tx.Rollback()
		if errRollback != nil {
			log.Error("failed to rollback", zap.Any("tableName", tableName), zap.Error(errRollback))
		}
		return errors.WrapError(errors.ErrMySQLTxnError, errors.WithMessage(err, fmt.Sprintf("create %s table: use %s db fail;", tableName, dbName)))
	}

	_, err = tx.Exec(createTableQuery)
	if err != nil {
		errRollback := tx.Rollback()
		if errRollback != nil {
			log.Error("failed to rollback", zap.Any("tableName", tableName), zap.Error(errRollback))
		}
		return errors.WrapError(errors.ErrMySQLTxnError, errors.WithMessage(err, fmt.Sprintf("create %s table: Exec fail; Query is %s", tableName, createTableQuery)))
	}
	err = tx.Commit()
	if err != nil {
		return errors.WrapError(errors.ErrMySQLTxnError, errors.WithMessage(err, fmt.Sprintf("create %s table: Commit Failed; Query is %s", tableName, createTableQuery)))
	}
	return nil
}

func (w *Writer) createDDLTsTable() error {
	database := filter.TiCDCSystemSchema
	query := `CREATE TABLE IF NOT EXISTS %s
	(
		ticdc_cluster_id varchar (255),
		changefeed varchar(255),
		ddl_ts varchar(18),
		table_id bigint(21),
		finished bool,
		is_syncpoint bool,
		INDEX (ticdc_cluster_id, changefeed, table_id),
		PRIMARY KEY (ticdc_cluster_id, changefeed, table_id)
	);`
	query = fmt.Sprintf(query, filter.DDLTsTable)

	return w.createTable(database, filter.DDLTsTable, query)
}

func (w *Writer) createDDLTsTableIfNotExist() error {
	w.ddlTsTableInitMutex.Lock()
	defer w.ddlTsTableInitMutex.Unlock()
	if w.ddlTsTableInit {
		return nil
	}
	// create checkpoint ts table if not exist
	err := w.createDDLTsTable()
	if err != nil {
		return err
	}
	w.ddlTsTableInit = true
	return nil
}