diff --git a/api/v2/changefeed_toml_test.go b/api/v2/changefeed_toml_test.go index 4bfdeee940..e901f7f17a 100644 --- a/api/v2/changefeed_toml_test.go +++ b/api/v2/changefeed_toml_test.go @@ -88,11 +88,16 @@ func TestChangeFeedInfoTOMLRoundTripToInternal(t *testing.T) { SinkURI: "blackhole://", StartTs: 449999999999999999, Config: &ReplicaConfig{ - PerformanceMode: util.AddressOf(config.PerformanceModeLowLatency), - MemoryQuota: util.AddressOf(uint64(1024)), - CaseSensitive: util.AddressOf(true), - ForceReplicate: util.AddressOf(true), - CheckGCSafePoint: util.AddressOf(false), + PerformanceMode: util.AddressOf(config.PerformanceModeLowLatency), + MemoryQuota: util.AddressOf(uint64(1024)), + CaseSensitive: util.AddressOf(true), + ForceReplicate: util.AddressOf(true), + CheckGCSafePoint: util.AddressOf(false), + Sink: &SinkConfig{ + DebeziumConfig: &DebeziumConfig{ + IncludeStartTs: util.AddressOf(true), + }, + }, SyncPointInterval: &JSONDuration{duration: 10 * time.Minute}, Integrity: &IntegrityConfig{ IntegrityCheckLevel: util.AddressOf("correctness"), @@ -114,6 +119,8 @@ func TestChangeFeedInfoTOMLRoundTripToInternal(t *testing.T) { // Top-level kebab-case keys and runtime field omissions. require.Contains(t, out, `sink-uri = "blackhole://"`) require.Contains(t, out, "start-ts") + require.Contains(t, out, "[config.sink.debezium]") + require.Contains(t, out, "include-start-ts = true") require.NotContains(t, out, "gid") // GID is omitted from TOML (toml:"-") // The [config] section must decode into the internal ReplicaConfig used by @@ -131,6 +138,7 @@ func TestChangeFeedInfoTOMLRoundTripToInternal(t *testing.T) { require.Equal(t, 10*time.Minute, *wrapper.Config.SyncPointInterval) require.Equal(t, "correctness", util.GetOrZero(wrapper.Config.Integrity.IntegrityCheckLevel)) require.Equal(t, "eventual", util.GetOrZero(wrapper.Config.Consistent.Level)) + require.True(t, util.GetOrZero(wrapper.Config.Sink.Debezium.IncludeStartTs)) } func TestCodecConfigTOMLRoundTripToInternal(t *testing.T) { diff --git a/api/v2/model.go b/api/v2/model.go index 9083c92a7d..4a4c4afbe9 100644 --- a/api/v2/model.go +++ b/api/v2/model.go @@ -546,6 +546,9 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig( debeziumConfig = &config.DebeziumConfig{ OutputOldValue: c.Sink.DebeziumConfig.OutputOldValue, } + if c.Sink.DebeziumConfig.IncludeStartTs != nil { + debeziumConfig.IncludeStartTs = util.AddressOf(*c.Sink.DebeziumConfig.IncludeStartTs) + } } var openProtocolConfig *config.OpenProtocolConfig if c.Sink.OpenProtocolConfig != nil { @@ -912,6 +915,9 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig { debeziumConfig = &DebeziumConfig{ OutputOldValue: cloned.Sink.Debezium.OutputOldValue, } + if cloned.Sink.Debezium.IncludeStartTs != nil { + debeziumConfig.IncludeStartTs = util.AddressOf(*cloned.Sink.Debezium.IncludeStartTs) + } } var openProtocolConfig *OpenProtocolConfig if cloned.Sink.OpenProtocol != nil { @@ -1591,7 +1597,8 @@ type OpenProtocolConfig struct { // DebeziumConfig represents the configurations for debezium protocol encoding type DebeziumConfig struct { - OutputOldValue bool `json:"output_old_value" toml:"output-old-value"` + OutputOldValue bool `json:"output_old_value" toml:"output-old-value"` + IncludeStartTs *bool `json:"include_start_ts,omitempty" toml:"include-start-ts,omitempty"` } type DispatcherCount struct { diff --git a/api/v2/model_test.go b/api/v2/model_test.go index ef48ced8be..17c9ae9093 100644 --- a/api/v2/model_test.go +++ b/api/v2/model_test.go @@ -42,6 +42,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), @@ -75,6 +78,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)) @@ -103,6 +107,7 @@ 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.Equal(t, 16, *apiCfgBack.Mounter.WorkerNum) require.True(t, *apiCfgBack.Scheduler.EnableTableAcrossNodes) require.Equal(t, "correctness", *apiCfgBack.Integrity.IntegrityCheckLevel) diff --git a/pkg/config/sink.go b/pkg/config/sink.go index 55645a073d..90632f8477 100644 --- a/pkg/config/sink.go +++ b/pkg/config/sink.go @@ -1169,6 +1169,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 dbb584a8ad..58c34aaab9 100644 --- a/pkg/sink/codec/common/config.go +++ b/pkg/sink/codec/common/config.go @@ -99,6 +99,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 } @@ -145,6 +148,7 @@ func NewConfig(protocol config.Protocol) *Config { DebeziumOutputOldValue: true, OpenOutputOldValue: true, DebeziumDisableSchema: false, + DebeziumIncludeStartTs: false, CSVOutputFieldHeader: false, } } @@ -185,7 +189,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"` @@ -203,6 +208,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 } @@ -313,6 +322,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 } @@ -345,6 +360,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 @@ -388,6 +406,12 @@ func (c *Config) Validate() error { ) } + 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 || c.Protocol == config.ProtocolDebeziumAvro { if c.AvroConfluentSchemaRegistry != "" && c.AvroGlueSchemaRegistry != nil { protocol := "Avro" diff --git a/pkg/sink/codec/common/config_test.go b/pkg/sink/codec/common/config_test.go index c85209638f..6247271f98 100644 --- a/pkg/sink/codec/common/config_test.go +++ b/pkg/sink/codec/common/config_test.go @@ -18,6 +18,7 @@ import ( "testing" "github.com/pingcap/ticdc/pkg/config" + "github.com/pingcap/ticdc/pkg/errors" "github.com/pingcap/ticdc/pkg/util" "github.com/stretchr/testify/require" ) @@ -53,3 +54,39 @@ func TestAvroIncludeBeforeValueConfigFile(t *testing.T) { require.False(t, cfg.EnableTiDBExtension) require.True(t, cfg.AvroIncludeBeforeValue) } + +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.ProtocolDebeziumAvro) + 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/avro.go b/pkg/sink/codec/debezium/avro.go index fcadacf00b..8ea205e9be 100644 --- a/pkg/sink/codec/debezium/avro.go +++ b/pkg/sink/codec/debezium/avro.go @@ -454,7 +454,7 @@ func (c *dbzCodec) buildDebeziumConnectSourceSchema( ) (*debeziumConnectSchema, error) { buf := &bytes.Buffer{} writer := util.BorrowJSONWriter(buf) - c.writeSourceSchema(writer, schemaName) + c.writeSourceSchema(writer, schemaName, false) util.ReturnJSONWriter(writer) return decodeDebeziumConnectSchema(buf.Bytes()) diff --git a/pkg/sink/codec/debezium/avro_test.go b/pkg/sink/codec/debezium/avro_test.go index a61e2094ed..524084cd7d 100644 --- a/pkg/sink/codec/debezium/avro_test.go +++ b/pkg/sink/codec/debezium/avro_test.go @@ -60,6 +60,8 @@ func TestDebeziumConfluentAvroEncodeRowEvent(t *testing.T) { cfg.AvroConfluentSchemaRegistry = "http://127.0.0.1:8081" cfg.AvroBigintUnsignedHandlingMode = common.BigintUnsignedHandlingModeString cfg.DebeziumDisableSchema = true + // debezium-include-start-ts must not affect the Avro protocol. + cfg.DebeziumIncludeStartTs = true cfg.TimeZone = time.UTC encoder, err := NewAvroBatchEncoder(ctx, cfg, "dbserver1") @@ -103,6 +105,9 @@ func TestDebeziumConfluentAvroEncodeRowEvent(t *testing.T) { require.Nil(t, source["snapshot"]) require.Nil(t, source["thread"]) require.Equal(t, "dbserver1", source["name"]) + // start_ts is a JSON-protocol-only field: the Avro payload and its + // registered schema must not carry it, even with debezium-include-start-ts on. + require.NotContains(t, source, "start_ts") valueSchema := decodeConfluentAvroSchemaForTest(t, messages[0].Value) require.Contains(t, valueSchema, `"name":"fooEnvelope"`) @@ -110,6 +115,7 @@ func TestDebeziumConfluentAvroEncodeRowEvent(t *testing.T) { require.Contains(t, valueSchema, `"name":"Source"`) require.Contains(t, valueSchema, `"logicalType":"decimal"`) require.NotContains(t, valueSchema, `"field":"transaction"`) + require.NotContains(t, valueSchema, `"field":"start_ts"`) } func TestDebeziumConfluentAvroSanitizesFullNameAndUnionBranch(t *testing.T) { diff --git a/pkg/sink/codec/debezium/codec.go b/pkg/sink/codec/debezium/codec.go index 1ceea135e8..f20a6fc136 100644 --- a/pkg/sink/codec/debezium/codec.go +++ b/pkg/sink/codec/debezium/codec.go @@ -886,7 +886,10 @@ func (c *dbzCodec) writeBinaryField(writer *util.JSONWriter, fieldName string, v writer.WriteBase64StringField(fieldName, value) } -func (c *dbzCodec) writeSourceSchema(writer *util.JSONWriter, schemaName string) { +// includeStartTs indicates whether start_ts should be declared in the source +// schema. DML callers pass the configured value, while DDL, checkpoint, and +// Avro callers pass false because their payloads do not carry the field. +func (c *dbzCodec) writeSourceSchema(writer *util.JSONWriter, schemaName string, includeStartTs bool) { writer.WriteObjectElement(func() { writer.WriteStringField("type", "struct") writer.WriteArrayField("fields", func() { @@ -987,6 +990,13 @@ func (c *dbzCodec) writeSourceSchema(writer *util.JSONWriter, schemaName string) writer.WriteStringField("field", "cluster_id") }) } + 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", c.sourceSchemaName(schemaName)) @@ -1084,6 +1094,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) }) @@ -1176,7 +1191,7 @@ func (c *dbzCodec) EncodeValue( jWriter.WriteRaw(fieldsJSON) }) }) - c.writeSourceSchema(jWriter, schemaName) + c.writeSourceSchema(jWriter, schemaName, c.config.DebeziumIncludeStartTs) jWriter.WriteObjectElement(func() { jWriter.WriteStringField("type", "string") jWriter.WriteBoolField("optional", false) @@ -1464,7 +1479,7 @@ func (c *dbzCodec) EncodeDDLEvent( jWriter.WriteIntField("version", 1) jWriter.WriteStringField("name", "io.debezium.connector.mysql.SchemaChangeValue") jWriter.WriteArrayField("fields", func() { - c.writeSourceSchema(jWriter, dbName) + c.writeSourceSchema(jWriter, dbName, false) jWriter.WriteObjectElement(func() { jWriter.WriteStringField("field", "ts_ms") jWriter.WriteBoolField("optional", false) @@ -1703,7 +1718,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, "watermark") + c.writeSourceSchema(jWriter, "watermark", 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 5b80cb8047..ac3a536b15 100644 --- a/pkg/sink/codec/debezium/decoder.go +++ b/pkg/sink/codec/debezium/decoder.go @@ -202,9 +202,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, @@ -250,6 +257,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) }