Skip to content

api: add ineligibleTables/eligible tables info when creating/updating changefeed #3932

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wk989898 wants to merge 8 commits into
base: master
Choose a base branch
from
+1,109 −889
Open

api: add ineligibleTables/eligible tables info when creating/updating changefeed #3932

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
39711f3
init
wk989898 Jan 5, 2026
a79c49a
add verbose
wk989898 Jan 12, 2026
3bd3af6
move table count metric to summary
wk989898 Jan 12, 2026
031a56a
update
wk989898 Jan 13, 2026
e957e9e
update
wk989898 Jan 13, 2026
89b94e9
Merge branch 'master' into changefeed-info
wk989898 Jan 13, 2026
66f7f99
update
wk989898 Jan 15, 2026
d9ca704
fix ut
wk989898 Jan 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions api/v2/api.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,8 @@ func RegisterOpenAPIV2Routes(router *gin.Engine, api OpenAPIV2) {

verifyTableGroup := v2.Group("/verify_table")
verifyTableGroup.POST("", api.VerifyTable)
getAllTablesGroup := v2.Group("/get_all_tables")
getAllTablesGroup.POST("", api.GetAllTables)

// processor apis
// Note: They are not useful in new arch cdc,
Expand Down
127 changes: 113 additions & 14 deletions api/v2/changefeed.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -221,7 +221,7 @@ func (h *OpenAPIV2) CreateChangefeed(c *gin.Context) {
return
}

ineligibleTables, _, err := getVerifiedTables(ctx, replicaCfg, kvStorage, cfg.StartTs, scheme, topic, protocol)
ineligibleTables, eligibleTables, err := getVerifiedTables(ctx, replicaCfg, kvStorage, cfg.StartTs, scheme, topic, protocol)
if err != nil {
_ = c.Error(err)
return
Expand Down Expand Up @@ -284,7 +284,10 @@ func (h *OpenAPIV2) CreateChangefeed(c *gin.Context) {
log.Info("Create changefeed successfully!",
zap.String("id", info.ChangefeedID.Name()),
zap.String("state", string(info.State)),
zap.String("changefeedInfo", info.String()))
zap.String("changefeedInfo", info.String()),
zap.Int("eligibleTablesLength", len(eligibleTables)),
zap.Int("ineligibleTablesLength", len(ineligibleTables)),
)

c.JSON(getStatus(c), CfInfoToAPIModel(
info,
Expand Down Expand Up @@ -352,7 +355,44 @@ func (h *OpenAPIV2) ListChangeFeeds(c *gin.Context) {
c.JSON(http.StatusOK, toListResponse(c, commonInfos))
}

// VerifyTable verify table, return ineligibleTables and EligibleTables.
// GetAllTables return ineligibleTables and EligibleTables.
func (h *OpenAPIV2) GetAllTables(c *gin.Context) {
ctx := c.Request.Context()
cfg := &ChangefeedConfig{ReplicaConfig: GetDefaultReplicaConfig()}

if err := c.BindJSON(&cfg); err != nil {
_ = c.Error(errors.WrapError(errors.ErrAPIInvalidParam, err))
return
}

// fill replicaConfig
replicaCfg := cfg.ReplicaConfig.ToInternalReplicaConfig()

keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
keyspaceName := GetKeyspaceValueWithDefault(c)
kvStorage, err := keyspaceManager.GetStorage(ctx, keyspaceName)
if err != nil {
_ = c.Error(err)
return
}

f, err := filter.NewFilter(replicaCfg.Filter, "", util.GetOrZero(replicaCfg.CaseSensitive), util.GetOrZero(replicaCfg.ForceReplicate))
if err != nil {
_ = c.Error(err)
return
}
_, ineligibleTables, eligibleTables, err := schemastore.
VerifyTables(f, kvStorage, cfg.StartTs)
if err != nil {
return
}
tables := &Tables{
IneligibleTables: toAPIModelFunc(ineligibleTables),
EligibleTables: toAPIModelFunc(eligibleTables),
}
c.JSON(http.StatusOK, tables)
}

func (h *OpenAPIV2) VerifyTable(c *gin.Context) {
ctx := c.Request.Context()
cfg := &ChangefeedConfig{ReplicaConfig: GetDefaultReplicaConfig()}
Expand Down Expand Up @@ -405,15 +445,6 @@ func (h *OpenAPIV2) VerifyTable(c *gin.Context) {
zap.Bool("ignoreIneligibleTable", util.GetOrZero(cfg.ReplicaConfig.IgnoreIneligibleTable)),
)

toAPIModelFunc := func(tbls []string) []TableName {
var apiModels []TableName
for _, tbl := range tbls {
apiModels = append(apiModels, TableName{
Table: tbl,
})
}
return apiModels
}
tables := &Tables{
IneligibleTables: toAPIModelFunc(ineligibleTables),
EligibleTables: toAPIModelFunc(eligibleTables),
Expand Down Expand Up @@ -466,6 +497,16 @@ func shouldShowRunningError(state config.FeedState) bool {
}
}

func toAPIModelFunc(tbls []string) []TableName {
var apiModels []TableName
for _, tbl := range tbls {
apiModels = append(apiModels, TableName{
Table: tbl,
})
}
return apiModels
}

func CfInfoToAPIModel(
info *config.ChangeFeedInfo,
status *config.ChangeFeedStatus,
Expand Down Expand Up @@ -666,8 +707,10 @@ func (h *OpenAPIV2) ResumeChangefeed(c *gin.Context) {

// If there is no overrideCheckpointTs, then check whether the currentCheckpointTs is smaller than gc safepoint or not.
newCheckpointTs := status.CheckpointTs
overwriteCheckpointTs := false
if cfg.OverwriteCheckpointTs != 0 {
newCheckpointTs = cfg.OverwriteCheckpointTs
overwriteCheckpointTs = true
}

keyspaceMeta := middleware.GetKeyspaceFromContext(c)
Expand Down Expand Up @@ -709,13 +752,55 @@ func (h *OpenAPIV2) ResumeChangefeed(c *gin.Context) {
}
}()

err = co.ResumeChangefeed(ctx, cfInfo.ChangefeedID, newCheckpointTs, cfg.OverwriteCheckpointTs != 0)
var (
eligibleTables []string
ineligibleTables []string
)
if overwriteCheckpointTs {
sinkURIParsed, err := url.Parse(cfInfo.SinkURI)
if err != nil {
_ = c.Error(errors.WrapError(errors.ErrSinkURIInvalid, err, cfInfo.SinkURI))
return
}
scheme := sinkURIParsed.Scheme
topic := ""
if config.IsMQScheme(scheme) {
topic, err = helper.GetTopic(sinkURIParsed)
if err != nil {
_ = c.Error(errors.WrapError(errors.ErrSinkURIInvalid, err, cfInfo.SinkURI))
return
}
}
protocol, _ := config.ParseSinkProtocolFromString(util.GetOrZero(cfInfo.Config.Sink.Protocol))

keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
kvStorage, err := keyspaceManager.GetStorage(ctx, keyspaceName)
if err != nil {
_ = c.Error(err)
return
}
ineligibleTables, eligibleTables, err = getVerifiedTables(ctx, cfInfo.Config, kvStorage, newCheckpointTs, scheme, topic, protocol)
if err != nil {
_ = c.Error(err)
return
}
}

err = co.ResumeChangefeed(ctx, cfInfo.ChangefeedID, newCheckpointTs, overwriteCheckpointTs)
if err != nil {
needRemoveGCSafePoint = true
_ = c.Error(err)
return
}
c.Errors = nil
log.Info("Resume changefeed successfully!",
zap.String("id", cfInfo.ChangefeedID.Name()),
zap.String("state", string(cfInfo.State)),
zap.String("changefeedInfo", cfInfo.String()),
zap.Bool("overwriteCheckpointTs", overwriteCheckpointTs),
zap.Int("eligibleTablesLength", len(eligibleTables)),
zap.Int("ineligibleTablesLength", len(ineligibleTables)),
)
c.JSON(getStatus(c), &EmptyResponse{})
}

Expand Down Expand Up @@ -813,6 +898,10 @@ func (h *OpenAPIV2) UpdateChangefeed(c *gin.Context) {
return
}

var (
ineligibleTables []string
eligibleTables []string
)
if configUpdated || sinkURIUpdated {
// verify replicaConfig
sinkURIParsed, err := url.Parse(oldCfInfo.SinkURI)
Expand Down Expand Up @@ -846,7 +935,7 @@ func (h *OpenAPIV2) UpdateChangefeed(c *gin.Context) {
}

// use checkpointTs get snapshot from kv storage
ineligibleTables, _, err := getVerifiedTables(ctx, oldCfInfo.Config, kvStorage, status.CheckpointTs, scheme, topic, protocol)
ineligibleTables, eligibleTables, err = getVerifiedTables(ctx, oldCfInfo.Config, kvStorage, status.CheckpointTs, scheme, topic, protocol)
if err != nil {
_ = c.Error(errors.ErrChangefeedUpdateRefused.GenWithStackByCause(err))
return
Expand All @@ -871,6 +960,16 @@ func (h *OpenAPIV2) UpdateChangefeed(c *gin.Context) {
return
}

log.Info("Update changefeed successfully!",
Copy link
Collaborator

@asddongmen asddongmen Jan 7, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please also add cost time in this log.

Copy link
Collaborator Author

@wk989898 wk989898 Jan 16, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The LogMiddleware will print the cost time

zap.String("id", oldCfInfo.ChangefeedID.Name()),
zap.String("state", string(oldCfInfo.State)),
zap.String("changefeedInfo", oldCfInfo.String()),
zap.Bool("configUpdated", configUpdated),
zap.Bool("sinkURIUpdated", sinkURIUpdated),
zap.Int("eligibleTablesLength", len(eligibleTables)),
zap.Int("ineligibleTablesLength", len(ineligibleTables)),
)

c.JSON(getStatus(c), CfInfoToAPIModel(oldCfInfo, status, nil))
}

Expand Down
8 changes: 7 additions & 1 deletion cmd/cdc/cli/cli_changefeed_create.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,6 +106,7 @@ type createChangefeedOptions struct {
disableGCSafePointCheck bool
startTs uint64
timezone string
verbose bool

cfg *config.ReplicaConfig
}
Expand All @@ -126,6 +127,7 @@ func (o *createChangefeedOptions) addFlags(cmd *cobra.Command) {
cmd.PersistentFlags().BoolVarP(&o.disableGCSafePointCheck, "disable-gc-check", "", false, "Disable GC safe point check")
cmd.PersistentFlags().Uint64Var(&o.startTs, "start-ts", 0, "Start ts of changefeed")
cmd.PersistentFlags().StringVar(&o.timezone, "tz", "SYSTEM", "timezone used when checking sink uri (changefeed timezone is determined by cdc server)")
cmd.PersistentFlags().BoolVar(&o.verbose, "verbose", false, "Print verbose information during creating changefeed")
// we don't support specify these flags below when cdc version >= 6.2.0
_ = cmd.PersistentFlags().MarkHidden("tz")
}
Expand Down Expand Up @@ -337,7 +339,11 @@ func (o *createChangefeedOptions) run(ctx context.Context, cmd *cobra.Command) e
if err != nil {
return err
}
cmd.Printf("Create changefeed successfully!\nID: %s\nInfo: %s\n", info.ID, infoStr)
cmd.Printf("Create changefeed successfully!\nID: %s\nInfo: %s\nIneligibleTablesCount: %d\nEligibleTablesCount: %d\n", info.ID, infoStr, len(tables.IneligibleTables), len(tables.EligibleTables))
if o.verbose {
cmd.Printf("EligibleTables: %v\n", tables.EligibleTables)
cmd.Printf("IneligibleTablesCount: %v\n", tables.IneligibleTables)
}
return nil
}

Expand Down
45 changes: 42 additions & 3 deletions cmd/cdc/cli/cli_changefeed_resume.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,15 +21,16 @@ import (
v2 "github.com/pingcap/ticdc/api/v2"
"github.com/pingcap/ticdc/cmd/cdc/factory"
"github.com/pingcap/ticdc/cmd/util"
apiv2client "github.com/pingcap/ticdc/pkg/api/v2"
apiClient "github.com/pingcap/ticdc/pkg/api/v2"
cerror "github.com/pingcap/ticdc/pkg/errors"
putil "github.com/pingcap/ticdc/pkg/util"
"github.com/spf13/cobra"
"github.com/tikv/client-go/v2/oracle"
)

// resumeChangefeedOptions defines flags for the `cli changefeed resume` command.
type resumeChangefeedOptions struct {
apiClient apiv2client.APIV2Interface
apiClient apiClient.APIV2Interface

changefeedID string
keyspace string
Expand All @@ -38,6 +39,7 @@ type resumeChangefeedOptions struct {
overwriteCheckpointTs string
currentTso *v2.Tso
checkpointTs uint64
verbose bool

upstreamPDAddrs string
upstreamCaPath string
Expand Down Expand Up @@ -66,6 +68,7 @@ func (o *resumeChangefeedOptions) addFlags(cmd *cobra.Command) {
"Certificate path for TLS connection to upstream")
cmd.PersistentFlags().StringVar(&o.upstreamKeyPath, "upstream-key", "",
"Private key path for TLS connection to upstream")
cmd.PersistentFlags().BoolVar(&o.verbose, "verbose", false, "Print verbose information during updating changefeed")
// we don't support specify there flags below when cdc version <= 6.3.0
_ = cmd.PersistentFlags().MarkHidden("upstream-pd")
_ = cmd.PersistentFlags().MarkHidden("upstream-ca")
Expand Down Expand Up @@ -204,8 +207,44 @@ func (o *resumeChangefeedOptions) run(cmd *cobra.Command) error {
if err := o.confirmResumeChangefeedCheck(cmd); err != nil {
return err
}
tables := &v2.Tables{}
if o.checkpointTs != 0 {
cf, err := o.apiClient.Changefeeds().Get(ctx, o.keyspace, o.changefeedID)
if err != nil {
return err
}
tables, err := o.apiClient.Changefeeds().GetAllTables(ctx, &v2.VerifyTableConfig{
ReplicaConfig: cf.Config,
StartTs: cf.CheckpointTs,
}, o.keyspace)
if err != nil {
return err
}
if len(tables.IneligibleTables) != 0 {
if putil.GetOrZero(cf.Config.ForceReplicate) {
cmd.Printf("[WARN] Force to replicate some ineligible tables, "+
"these tables do not have a primary key or a not-null unique key: %#v\n"+
"[WARN] This may cause data redundancy, "+
"please refer to the official documentation for details.\n",
tables.IneligibleTables)
} else {
cmd.Printf("[WARN] Some tables are not eligible to replicate, "+
"because they do not have a primary key or a not-null unique key: %#v\n",
tables.IneligibleTables)
}
}
}
err := o.apiClient.Changefeeds().Resume(ctx, cfg, o.keyspace, o.changefeedID)

if err != nil {
return err
}
// o.checkpointTs != 0
cmd.Printf("Resume changefeed successfully! "+
"\nID: %s\nOverwriteCheckpointTs: %t\nIneligibleTablesCount: %d\nEligibleTablesCount: %d\n", o.changefeedID, o.checkpointTs != 0, len(tables.IneligibleTables), len(tables.EligibleTables))
if o.verbose {
cmd.Printf("EligibleTables: %v\n", tables.EligibleTables)
cmd.Printf("IneligibleTablesCount: %v\n", tables.IneligibleTables)
}
return err
}

Expand Down
Loading