diff --git a/api/v2/model.go b/api/v2/model.go index 26a9c30e3c..8c23b2fee2 100644 --- a/api/v2/model.go +++ b/api/v2/model.go @@ -506,8 +506,16 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig( } var debeziumConfig *config.DebeziumConfig if c.Sink.DebeziumConfig != nil { + // Fall back to the default when OutputOldValue is omitted. + outputOldValue := config.DefaultDebeziumOutputOldValue + if c.Sink.DebeziumConfig.OutputOldValue != nil { + outputOldValue = *c.Sink.DebeziumConfig.OutputOldValue + } debeziumConfig = &config.DebeziumConfig{ - OutputOldValue: c.Sink.DebeziumConfig.OutputOldValue, + OutputOldValue: outputOldValue, + } + if c.Sink.DebeziumConfig.IncludeStartTs != nil { + debeziumConfig.IncludeStartTs = util.AddressOf(*c.Sink.DebeziumConfig.IncludeStartTs) } } var openProtocolConfig *config.OpenProtocolConfig @@ -863,7 +871,10 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig { var debeziumConfig *DebeziumConfig if cloned.Sink.Debezium != nil { debeziumConfig = &DebeziumConfig{ - OutputOldValue: cloned.Sink.Debezium.OutputOldValue, + OutputOldValue: util.AddressOf(cloned.Sink.Debezium.OutputOldValue), + } + if cloned.Sink.Debezium.IncludeStartTs != nil { + debeziumConfig.IncludeStartTs = util.AddressOf(*cloned.Sink.Debezium.IncludeStartTs) } } var openProtocolConfig *OpenProtocolConfig @@ -1545,7 +1556,8 @@ type OpenProtocolConfig struct { // DebeziumConfig represents the configurations for debezium protocol encoding type DebeziumConfig struct { - OutputOldValue bool `json:"output_old_value"` + OutputOldValue *bool `json:"output_old_value,omitempty"` + IncludeStartTs *bool `json:"include_start_ts,omitempty"` } type DispatcherCount struct { diff --git a/api/v2/model_test.go b/api/v2/model_test.go index 88fb821cdb..075f15e56f 100644 --- a/api/v2/model_test.go +++ b/api/v2/model_test.go @@ -41,6 +41,9 @@ func TestReplicaConfigConversion(t *testing.T) { SpoolDiskQuota: util.AddressOf(int64(1024)), SpoolBaseDir: util.AddressOf("/tmp/ticdc-spool"), }, + DebeziumConfig: &DebeziumConfig{ + IncludeStartTs: util.AddressOf(true), + }, }, Mounter: &MounterConfig{ WorkerNum: util.AddressOf(16), @@ -73,6 +76,7 @@ func TestReplicaConfigConversion(t *testing.T) { require.True(t, util.GetOrZero(internalCfg.Sink.CloudStorageConfig.UseTableIDAsPath)) require.Equal(t, int64(1024), util.GetOrZero(internalCfg.Sink.CloudStorageConfig.SpoolDiskQuota)) require.Equal(t, "/tmp/ticdc-spool", util.GetOrZero(internalCfg.Sink.CloudStorageConfig.SpoolBaseDir)) + require.True(t, util.GetOrZero(internalCfg.Sink.Debezium.IncludeStartTs)) require.Equal(t, internalCfg.Mounter.WorkerNum, *apiCfg.Mounter.WorkerNum) require.True(t, util.GetOrZero(internalCfg.Scheduler.EnableTableAcrossNodes)) require.Equal(t, 1000, util.GetOrZero(internalCfg.Scheduler.RegionThreshold)) @@ -82,6 +86,21 @@ func TestReplicaConfigConversion(t *testing.T) { require.Equal(t, int64(128), util.GetOrZero(internalCfg.Consistent.MaxLogSize)) require.Equal(t, int64(2000), util.GetOrZero(internalCfg.Consistent.FlushIntervalInMs)) require.Equal(t, "s3://test", util.GetOrZero(internalCfg.Consistent.Storage)) + // output_old_value is omitted in apiCfg and must keep its default (true). + require.True(t, internalCfg.Sink.Debezium.OutputOldValue) + + // An explicit output_old_value must be honored. + apiCfgDebezium := &ReplicaConfig{ + Sink: &SinkConfig{ + DebeziumConfig: &DebeziumConfig{ + OutputOldValue: util.AddressOf(false), + IncludeStartTs: util.AddressOf(true), + }, + }, + } + internalDebezium := apiCfgDebezium.ToInternalReplicaConfig() + require.False(t, internalDebezium.Sink.Debezium.OutputOldValue) + require.True(t, util.GetOrZero(internalDebezium.Sink.Debezium.IncludeStartTs)) // Test case 2: Nil fields (should use defaults or be nil) apiCfgNil := &ReplicaConfig{} @@ -100,6 +119,8 @@ func TestReplicaConfigConversion(t *testing.T) { require.True(t, *apiCfgBack.Sink.CloudStorageConfig.UseTableIDAsPath) require.Equal(t, int64(1024), *apiCfgBack.Sink.CloudStorageConfig.SpoolDiskQuota) require.Equal(t, "/tmp/ticdc-spool", *apiCfgBack.Sink.CloudStorageConfig.SpoolBaseDir) + require.True(t, util.GetOrZero(apiCfgBack.Sink.DebeziumConfig.IncludeStartTs)) + require.True(t, util.GetOrZero(apiCfgBack.Sink.DebeziumConfig.OutputOldValue)) require.Equal(t, 16, *apiCfgBack.Mounter.WorkerNum) require.True(t, *apiCfgBack.Scheduler.EnableTableAcrossNodes) require.Equal(t, "correctness", *apiCfgBack.Integrity.IntegrityCheckLevel) diff --git a/pkg/config/replica_config.go b/pkg/config/replica_config.go index b24df890bc..5590799e69 100644 --- a/pkg/config/replica_config.go +++ b/pkg/config/replica_config.go @@ -79,7 +79,7 @@ var defaultReplicaConfig = &ReplicaConfig{ SendAllBootstrapAtStart: util.AddressOf(DefaultSendAllBootstrapAtStart), DebeziumDisableSchema: util.AddressOf(false), OpenProtocol: &OpenProtocolConfig{OutputOldValue: true}, - Debezium: &DebeziumConfig{OutputOldValue: true}, + Debezium: &DebeziumConfig{OutputOldValue: DefaultDebeziumOutputOldValue}, }, Consistent: &ConsistentConfig{ Level: util.AddressOf("none"), diff --git a/pkg/config/sink.go b/pkg/config/sink.go index e0edb288ef..86de7e5d60 100644 --- a/pkg/config/sink.go +++ b/pkg/config/sink.go @@ -95,6 +95,10 @@ const ( // to send all tables bootstrap message at changefeed start. DefaultSendAllBootstrapAtStart = false + // DefaultDebeziumOutputOldValue is the default value of whether + // to output the old value in debezium protocol messages. + DefaultDebeziumOutputOldValue = true + // DefaultMaxReconnectToPulsarBroker is the default max reconnect times to pulsar broker. // The pulsar client uses an exponential backoff with jitter to reconnect to the broker. // Based on test, when the max reconnect times is 3, @@ -1166,6 +1170,9 @@ type OpenProtocolConfig struct { // DebeziumConfig represents the configurations for debezium protocol encoding type DebeziumConfig struct { OutputOldValue bool `toml:"output-old-value" json:"output-old-value"` + // IncludeStartTs controls whether the transaction start_ts is included in + // the source block of Debezium JSON output. + IncludeStartTs *bool `toml:"include-start-ts" json:"include-start-ts,omitempty"` } // validRoutingExpressionRegexp accepts routing expressions made of literal text diff --git a/pkg/sink/codec/common/config.go b/pkg/sink/codec/common/config.go index c55c30fac0..9ef42339d8 100644 --- a/pkg/sink/codec/common/config.go +++ b/pkg/sink/codec/common/config.go @@ -94,6 +94,9 @@ type Config struct { DebeziumDisableSchema bool // Debezium only. Whether before value should be included in the output. DebeziumOutputOldValue bool + // Debezium only. Whether the transaction start_ts should be included in + // the source block of the output. JSON protocol only. + DebeziumIncludeStartTs bool // CSV only. Whether header should be included in the output. CSVOutputFieldHeader bool } @@ -138,6 +141,7 @@ func NewConfig(protocol config.Protocol) *Config { DebeziumOutputOldValue: true, OpenOutputOldValue: true, DebeziumDisableSchema: false, + DebeziumIncludeStartTs: false, CSVOutputFieldHeader: false, } } @@ -177,7 +181,8 @@ type urlConfig struct { OnlyOutputUpdatedColumns *bool `form:"only-output-updated-columns"` ContentCompatible *bool `form:"content-compatible"` - DebeziumDisableSchema *bool `form:"debezium-disable-schema"` + DebeziumDisableSchema *bool `form:"debezium-disable-schema"` + DebeziumIncludeStartTs *bool `form:"debezium-include-start-ts"` // EncodingFormatType is only works for the simple protocol, // can be `json` and `avro`, default to `json`. EncodingFormatType *string `form:"encoding-format"` @@ -195,6 +200,10 @@ func (c *Config) Apply(sinkURI *url.URL, sinkConfig *config.SinkConfig) error { if err = binding.Query.Bind(req, urlParameter); err != nil { return errors.WrapError(errors.ErrSinkInvalidConfig, err) } + // Keep the raw URI parameters: mergeConfig uses mergo, which cannot + // override a *bool "true" (from the config file) with an explicit + // "false" from the sink URI, so explicit URI values are applied last. + rawURLParameter := urlParameter if urlParameter, err = mergeConfig(sinkConfig, urlParameter); err != nil { return err } @@ -300,6 +309,12 @@ func (c *Config) Apply(sinkURI *url.URL, sinkConfig *config.SinkConfig) error { if urlParameter.DebeziumDisableSchema != nil { c.DebeziumDisableSchema = *urlParameter.DebeziumDisableSchema } + if urlParameter.DebeziumIncludeStartTs != nil { + c.DebeziumIncludeStartTs = *urlParameter.DebeziumIncludeStartTs + } + if rawURLParameter.DebeziumIncludeStartTs != nil { + c.DebeziumIncludeStartTs = *rawURLParameter.DebeziumIncludeStartTs + } return nil } @@ -331,6 +346,9 @@ func mergeConfig( if sinkConfig.DebeziumDisableSchema != nil { dest.DebeziumDisableSchema = sinkConfig.DebeziumDisableSchema } + if sinkConfig.Debezium != nil && sinkConfig.Debezium.IncludeStartTs != nil { + dest.DebeziumIncludeStartTs = sinkConfig.Debezium.IncludeStartTs + } } if err := mergo.Merge(dest, urlParameters, mergo.WithOverride); err != nil { return nil, err @@ -360,6 +378,12 @@ func (c *Config) Validate() error { zap.String("protocol", c.Protocol.String())) } + if c.DebeziumIncludeStartTs && c.Protocol != config.ProtocolDebezium { + return errors.ErrCodecInvalidConfig.GenWithStack( + `debezium-include-start-ts only takes effect with protocol "debezium"`, + ) + } + if c.Protocol == config.ProtocolAvro { if c.AvroConfluentSchemaRegistry != "" && c.AvroGlueSchemaRegistry != nil { return errors.ErrCodecInvalidConfig.GenWithStack( diff --git a/pkg/sink/codec/common/config_test.go b/pkg/sink/codec/common/config_test.go index 4824bbfaec..847dd083b6 100644 --- a/pkg/sink/codec/common/config_test.go +++ b/pkg/sink/codec/common/config_test.go @@ -68,3 +68,39 @@ func TestValidateMessageLimits(t *testing.T) { }) } } + +func TestDebeziumIncludeStartTsConfig(t *testing.T) { + // URI parameter + cfg := NewConfig(config.ProtocolDebezium) + sinkURI, err := url.Parse("kafka://127.0.0.1:9092/topic?protocol=debezium&debezium-include-start-ts=true") + require.NoError(t, err) + require.NoError(t, cfg.Apply(sinkURI, config.GetDefaultReplicaConfig().Sink)) + require.True(t, cfg.DebeziumIncludeStartTs) + require.NoError(t, cfg.Validate()) + + // changefeed config file + on := true + cfg2 := NewConfig(config.ProtocolDebezium) + sinkConfig := config.GetDefaultReplicaConfig().Sink + sinkConfig.Debezium.IncludeStartTs = &on + sinkURI2, err := url.Parse("kafka://127.0.0.1:9092/topic?protocol=debezium") + require.NoError(t, err) + require.NoError(t, cfg2.Apply(sinkURI2, sinkConfig)) + require.True(t, cfg2.DebeziumIncludeStartTs) + + // URI parameter overrides the config file + cfg3 := NewConfig(config.ProtocolDebezium) + sinkConfig3 := config.GetDefaultReplicaConfig().Sink + sinkConfig3.Debezium.IncludeStartTs = &on + sinkURI3, err := url.Parse("kafka://127.0.0.1:9092/topic?protocol=debezium&debezium-include-start-ts=false") + require.NoError(t, err) + require.NoError(t, cfg3.Apply(sinkURI3, sinkConfig3)) + require.False(t, cfg3.DebeziumIncludeStartTs) + + // only supported by the debezium (JSON) protocol + cfg4 := NewConfig(config.ProtocolCanalJSON) + cfg4.DebeziumIncludeStartTs = true + errCode, ok := errors.RFCCode(cfg4.Validate()) + require.True(t, ok) + require.Equal(t, errors.ErrCodecInvalidConfig.RFCCode(), errCode) +} diff --git a/pkg/sink/codec/debezium/codec.go b/pkg/sink/codec/debezium/codec.go index 9f7853bb33..fc47d8d2d4 100644 --- a/pkg/sink/codec/debezium/codec.go +++ b/pkg/sink/codec/debezium/codec.go @@ -758,7 +758,10 @@ func (c *dbzCodec) writeBinaryField(writer *util.JSONWriter, fieldName string, v writer.WriteBase64StringField(fieldName, value) } -func (c *dbzCodec) writeSourceSchema(writer *util.JSONWriter) { +// includeStartTs indicates whether start_ts should be declared in the source +// schema. DML callers pass the configured value, while DDL and checkpoint +// callers pass false because their payloads do not carry the field. +func (c *dbzCodec) writeSourceSchema(writer *util.JSONWriter, includeStartTs bool) { writer.WriteObjectElement(func() { writer.WriteStringField("type", "struct") writer.WriteArrayField("fields", func() { @@ -843,6 +846,13 @@ func (c *dbzCodec) writeSourceSchema(writer *util.JSONWriter) { writer.WriteBoolField("optional", true) writer.WriteStringField("field", "query") }) + if includeStartTs { + writer.WriteObjectElement(func() { + writer.WriteStringField("type", "int64") + writer.WriteBoolField("optional", false) + writer.WriteStringField("field", "start_ts") + }) + } }) writer.WriteBoolField("optional", false) writer.WriteStringField("name", "io.debezium.connector.mysql.Source") @@ -933,6 +943,11 @@ func (c *dbzCodec) EncodeValue( // The followings are TiDB extended fields jWriter.WriteUint64Field("commit_ts", e.CommitTs) + // start_ts: the start TSO of the transaction that made this change, + // exposed for downstream consumers that need transaction correlation. + if c.config.DebeziumIncludeStartTs { + jWriter.WriteUint64Field("start_ts", e.StartTs) + } jWriter.WriteStringField("cluster_id", c.clusterID) }) @@ -1026,7 +1041,7 @@ func (c *dbzCodec) EncodeValue( jWriter.WriteRaw(fieldsJSON) }) }) - c.writeSourceSchema(jWriter) + c.writeSourceSchema(jWriter, c.config.DebeziumIncludeStartTs) jWriter.WriteObjectElement(func() { jWriter.WriteStringField("type", "string") jWriter.WriteBoolField("optional", false) @@ -1312,7 +1327,7 @@ func (c *dbzCodec) EncodeDDLEvent( jWriter.WriteIntField("version", 1) jWriter.WriteStringField("name", "io.debezium.connector.mysql.SchemaChangeValue") jWriter.WriteArrayField("fields", func() { - c.writeSourceSchema(jWriter) + c.writeSourceSchema(jWriter, false) jWriter.WriteObjectElement(func() { jWriter.WriteStringField("field", "ts_ms") jWriter.WriteBoolField("optional", false) @@ -1551,7 +1566,7 @@ func (c *dbzCodec) EncodeCheckpointEvent( fmt.Sprintf("%s.%s.Envelope", common.SanitizeName(c.clusterID), "watermark")) jWriter.WriteIntField("version", 1) jWriter.WriteArrayField("fields", func() { - c.writeSourceSchema(jWriter) + c.writeSourceSchema(jWriter, false) jWriter.WriteObjectElement(func() { jWriter.WriteStringField("type", "string") jWriter.WriteBoolField("optional", false) diff --git a/pkg/sink/codec/debezium/codec_test.go b/pkg/sink/codec/debezium/codec_test.go index b23167d37e..67d9f0e97d 100644 --- a/pkg/sink/codec/debezium/codec_test.go +++ b/pkg/sink/codec/debezium/codec_test.go @@ -1544,3 +1544,44 @@ func BenchmarkEncodeLargeBinary(b *testing.B) { codec.EncodeValue(e, buf) } } + +func TestStartTsNotInDDLAndCheckpointEvents(t *testing.T) { + // Even with debezium-include-start-ts enabled, DDL and checkpoint + // (watermark) messages must not declare start_ts in their schemas: + // their payloads never carry the field (no per-row transaction), and a + // declared-but-absent non-optional field breaks schema-validating consumers. + codec := &dbzCodec{ + config: common.NewConfig(config.ProtocolDebezium), + clusterID: "test_cluster", + nowFunc: func() time.Time { return time.Unix(1701326309, 0) }, + } + codec.config.DebeziumIncludeStartTs = true + codec.config.DebeziumDisableSchema = false + + helper := commonEvent.NewEventTestHelper(t) + defer helper.Close() + helper.Tk().MustExec("use test") + helper.DDL2Job(`create table test.table1(id int(10) primary key)`) + job := helper.DDL2Job(`RENAME TABLE test.table1 to test.table2`) + tableInfo := helper.GetTableInfo(job) + + e := &commonEvent.DDLEvent{ + FinishedTs: 1, + TableInfo: tableInfo, + SchemaName: "test", + TableName: "table2", + ExtraSchemaName: "test", + ExtraTableName: "table1", + Type: byte(timodel.ActionRenameTable), + Query: job.Query, + } + keyBuf := bytes.NewBuffer(nil) + buf := bytes.NewBuffer(nil) + require.NoError(t, codec.EncodeDDLEvent(e, keyBuf, buf)) + require.NotContains(t, buf.String(), "start_ts") + + keyBuf.Reset() + buf.Reset() + require.NoError(t, codec.EncodeCheckpointEvent(3, keyBuf, buf)) + require.NotContains(t, buf.String(), "start_ts") +} diff --git a/pkg/sink/codec/debezium/debezium_test.go b/pkg/sink/codec/debezium/debezium_test.go index 80380bfebf..44cefdb9af 100644 --- a/pkg/sink/codec/debezium/debezium_test.go +++ b/pkg/sink/codec/debezium/debezium_test.go @@ -14,6 +14,7 @@ package debezium import ( + "bytes" "context" "encoding/json" "os" @@ -207,3 +208,152 @@ func (s *debeziumSuite) TestDataTypes() { s.requireDebeziumJSONEq(dataDbzOutput, messages[0].Value) s.requireDebeziumJSONEq(keyDbzOutput, messages[0].Key) } + +func TestEncodeStartTsInSource(t *testing.T) { + // The field is emitted when debezium-include-start-ts is enabled, + // independent of enable-tidb-extension. + cfg := common.NewConfig(config.ProtocolDebezium) + cfg.DebeziumIncludeStartTs = true + cfg.TimeZone = time.UTC + + encoder := NewBatchEncoder(cfg, "dbserver1") + rowEvent := common.NewRoutedRowEvent4Test() + rowEvent.StartTs = 5 + require.NoError(t, encoder.AppendRowChangedEvent(context.Background(), "", rowEvent)) + + messages := encoder.Build() + require.Len(t, messages, 1) + + dec := json.NewDecoder(bytes.NewReader(messages[0].Value)) + dec.UseNumber() + var value map[string]any + require.NoError(t, dec.Decode(&value)) + payload := value["payload"].(map[string]any) + source := payload["source"].(map[string]any) + startTs, err := source["start_ts"].(json.Number).Int64() + require.NoError(t, err) + require.Equal(t, int64(5), startTs) + + // The source schema declares start_ts under the same switch, so + // schema-validated consumers can see it without enable-tidb-extension. + schema := value["schema"].(map[string]any) + sourceSchema := schemaFieldsByName(t, schema, "source") + require.NotNil(t, sourceSchema) + require.NotNil(t, schemaFieldsByName(t, sourceSchema, "start_ts")) + + // round-trip: decoding restores the true start ts. The TiCDC-side decoder + // requires enable-tidb-extension: it relies on the per-column tidb_type in + // the schema to reconstruct column types, so the encoded message must + // carry the extension fields as well. + cfg2 := common.NewConfig(config.ProtocolDebezium) + cfg2.DebeziumIncludeStartTs = true + cfg2.EnableTiDBExtension = true + cfg2.TimeZone = time.UTC + encoder2 := NewBatchEncoder(cfg2, "dbserver1") + require.NoError(t, encoder2.AppendRowChangedEvent(context.Background(), "", rowEvent)) + messages2 := encoder2.Build() + require.Len(t, messages2, 1) + + decoder := NewDecoder(cfg2, 0, nil) + decoder.AddKeyValue(messages2[0].Key, messages2[0].Value) + messageType, hasNext := decoder.HasNext() + require.True(t, hasNext) + require.Equal(t, common.MessageTypeRow, messageType) + decoded := decoder.NextDMLMessage().ToDMLEvent() + require.Equal(t, uint64(5), decoded.GetStartTs()) +} + +func TestDecodeStartTsFallbackToCommitTs(t *testing.T) { + // A message produced without debezium-include-start-ts (the pre-feature + // format) has no start_ts in the source block; decoding it must fall back + // to commit_ts, keeping the old behavior. + cfg := common.NewConfig(config.ProtocolDebezium) + cfg.EnableTiDBExtension = true // required to decode the message back + cfg.TimeZone = time.UTC + + encoder := NewBatchEncoder(cfg, "dbserver1") + rowEvent := common.NewRoutedRowEvent4Test() + rowEvent.StartTs = 5 + require.NoError(t, encoder.AppendRowChangedEvent(context.Background(), "", rowEvent)) + + messages := encoder.Build() + require.Len(t, messages, 1) + + dec := json.NewDecoder(bytes.NewReader(messages[0].Value)) + dec.UseNumber() + var value map[string]any + require.NoError(t, dec.Decode(&value)) + payload := value["payload"].(map[string]any) + source := payload["source"].(map[string]any) + require.NotContains(t, source, "start_ts") + schema := value["schema"].(map[string]any) + sourceSchema := schemaFieldsByName(t, schema, "source") + require.NotNil(t, sourceSchema) + require.Nil(t, schemaFieldsByName(t, sourceSchema, "start_ts")) + + decoder := NewDecoder(cfg, 0, nil) + decoder.AddKeyValue(messages[0].Key, messages[0].Value) + messageType, hasNext := decoder.HasNext() + require.True(t, hasNext) + require.Equal(t, common.MessageTypeRow, messageType) + decoded := decoder.NextDMLMessage().ToDMLEvent() + require.Equal(t, decoded.GetCommitTs(), decoded.GetStartTs()) + require.NotEqual(t, uint64(5), decoded.GetStartTs()) +} + +func TestDecodeNonPositiveStartTsFallbackToCommitTs(t *testing.T) { + cfg := common.NewConfig(config.ProtocolDebezium) + cfg.DebeziumIncludeStartTs = true + cfg.EnableTiDBExtension = true + cfg.TimeZone = time.UTC + + encoder := NewBatchEncoder(cfg, "dbserver1") + rowEvent := common.NewRoutedRowEvent4Test() + rowEvent.StartTs = 5 + require.NoError(t, encoder.AppendRowChangedEvent(context.Background(), "", rowEvent)) + + messages := encoder.Build() + require.Len(t, messages, 1) + + for _, tc := range []struct { + name string + startTs json.Number + }{ + {name: "zero", startTs: json.Number("0")}, + {name: "negative", startTs: json.Number("-1")}, + } { + t.Run(tc.name, func(t *testing.T) { + dec := json.NewDecoder(bytes.NewReader(messages[0].Value)) + dec.UseNumber() + var value map[string]any + require.NoError(t, dec.Decode(&value)) + payload := value["payload"].(map[string]any) + source := payload["source"].(map[string]any) + source["start_ts"] = tc.startTs + valueBytes, err := json.Marshal(value) + require.NoError(t, err) + + decoder := NewDecoder(cfg, 0, nil) + decoder.AddKeyValue(messages[0].Key, valueBytes) + messageType, hasNext := decoder.HasNext() + require.True(t, hasNext) + require.Equal(t, common.MessageTypeRow, messageType) + decoded := decoder.NextDMLMessage().ToDMLEvent() + require.Equal(t, decoded.GetCommitTs(), decoded.GetStartTs()) + }) + } +} + +// schemaFieldsByName returns the sub-schema object of a field inside a Debezium +// struct schema, or nil when the field is not declared. +func schemaFieldsByName(t *testing.T, schema map[string]any, name string) map[string]any { + fields, ok := schema["fields"].([]any) + require.True(t, ok) + for _, f := range fields { + fm := f.(map[string]any) + if fm["field"] == name { + return fm + } + } + return nil +} diff --git a/pkg/sink/codec/debezium/decoder.go b/pkg/sink/codec/debezium/decoder.go index 2d2ae1675d..cb4f9bc9a8 100644 --- a/pkg/sink/codec/debezium/decoder.go +++ b/pkg/sink/codec/debezium/decoder.go @@ -201,9 +201,16 @@ func (d *decoder) assembleDMLEventFromPayload( ) *commonEvent.DMLEvent { tableInfo := queryTableInfoFromPayload(keyPayload, valuePayload, valueSchema) commitTs := getCommitTsFromPayload(valuePayload) + startTs, hasStartTs := getStartTsFromPayload(valuePayload) + if !hasStartTs { + // Keep old messages consumable when start_ts is absent. Invalid values + // are logged by getStartTsFromPayload and also fall back so a malformed + // message does not stop production consumption. + startTs = commitTs + } event := &commonEvent.DMLEvent{ Rows: chunk.NewChunkFromPoolWithCapacity(tableInfo.GetFieldSlice(), chunk.InitialCapacity), - StartTs: commitTs, + StartTs: startTs, CommitTs: commitTs, TableInfo: tableInfo, PhysicalTableID: tableInfo.TableName.TableID, @@ -249,6 +256,33 @@ func getCommitTsFromPayload(valuePayload map[string]any) uint64 { return uint64(commitTs) } +// getStartTsFromPayload returns the start_ts carried in the source block. +// It returns false when the field is absent or invalid. Invalid values are +// logged before returning so callers can fall back without stopping consumption. +func getStartTsFromPayload(valuePayload map[string]any) (uint64, bool) { + source := valuePayload["source"].(map[string]any) + rawStartTs, exists := source["start_ts"] + if !exists { + return 0, false + } + startTs, ok := rawStartTs.(json.Number) + if !ok { + log.Error("decode value failed", + zap.String("reason", "start_ts is not an integer"), + zap.String("value", util.RedactAny(source))) + return 0, false + } + ts, err := startTs.Int64() + if err == nil && ts <= 0 { + err = errors.Errorf("start_ts must be positive: %d", ts) + } + if err != nil { + log.Error("decode value failed", zap.Error(err), zap.String("value", util.RedactAny(source))) + return 0, false + } + return uint64(ts), true +} + func (d *decoder) getSchemaName() string { return getSchemaNameFromPayload(d.valuePayload) }