debezium(ticdc): cli support debezium-disable-schema (#10372)

close #10371
pingcap · Dec 29, 2023 · 1931873 · 1931873
1 parent fdef38b
commit 1931873
Show file tree

Hide file tree

Showing 9 changed files with 24 additions and 3 deletions.
diff --git a/cdc/api/v2/model.go b/cdc/api/v2/model.go
@@ -486,6 +486,9 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig(
 		if c.Sink.AdvanceTimeoutInSec != nil {
 			res.Sink.AdvanceTimeoutInSec = util.AddressOf(*c.Sink.AdvanceTimeoutInSec)
 		}
+		if c.Sink.DebeziumDisableSchema != nil {
+			res.Sink.DebeziumDisableSchema = util.AddressOf(*c.Sink.DebeziumDisableSchema)
+		}
 
 	}
 	if c.Mounter != nil {
@@ -776,6 +779,9 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig {
 		if cloned.Sink.AdvanceTimeoutInSec != nil {
 			res.Sink.AdvanceTimeoutInSec = util.AddressOf(*cloned.Sink.AdvanceTimeoutInSec)
 		}
+		if cloned.Sink.DebeziumDisableSchema != nil {
+			res.Sink.DebeziumDisableSchema = util.AddressOf(*cloned.Sink.DebeziumDisableSchema)
+		}
 	}
 	if cloned.Consistent != nil {
 		res.Consistent = &ConsistentConfig{
@@ -950,6 +956,7 @@ type SinkConfig struct {
 	MySQLConfig                      *MySQLConfig        `json:"mysql_config,omitempty"`
 	CloudStorageConfig               *CloudStorageConfig `json:"cloud_storage_config,omitempty"`
 	AdvanceTimeoutInSec              *uint               `json:"advance_timeout,omitempty"`
+	DebeziumDisableSchema            *bool               `json:"debezium_disable_schema,omitempty"`
 }
 
 // CSVConfig denotes the csv config

diff --git a/cdc/api/v2/model_test.go b/cdc/api/v2/model_test.go
@@ -59,6 +59,7 @@ var defaultAPIConfig = &ReplicaConfig{
 		DeleteOnlyOutputHandleKeyColumns: util.AddressOf(false),
 		ContentCompatible:                util.AddressOf(false),
 		AdvanceTimeoutInSec:              util.AddressOf(uint(150)),
+		DebeziumDisableSchema:            util.AddressOf(false),
 	},
 	Consistent: &ConsistentConfig{
 		Level:                 "none",

diff --git a/pkg/cmd/util/helper_test.go b/pkg/cmd/util/helper_test.go
@@ -215,6 +215,7 @@ func TestAndWriteExampleReplicaTOML(t *testing.T) {
 		ContentCompatible:                util.AddressOf(false),
 		Protocol:                         util.AddressOf("open-protocol"),
 		AdvanceTimeoutInSec:              util.AddressOf(uint(150)),
+		DebeziumDisableSchema:            util.AddressOf(false),
 	}, cfg.Sink)
 }
 
@@ -249,6 +250,7 @@ func TestAndWriteStorageSinkTOML(t *testing.T) {
 		DeleteOnlyOutputHandleKeyColumns: util.AddressOf(false),
 		ContentCompatible:                util.AddressOf(false),
 		AdvanceTimeoutInSec:              util.AddressOf(uint(150)),
+		DebeziumDisableSchema:            util.AddressOf(false),
 	}, cfg.Sink)
 }
 

diff --git a/pkg/config/config_test_data.go b/pkg/config/config_test_data.go
@@ -65,7 +65,8 @@ const (
       "large-message-handle-compression": "",
       "claim-check-storage-uri": ""
     },
-    "advance-timeout-in-sec": 150
+    "advance-timeout-in-sec": 150,
+    "debezium-disable-schema": false
   },
   "consistent": {
     "level": "none",
@@ -318,7 +319,8 @@ const (
       "file-size": 1024,
       "output-column-id":false
     },
-    "advance-timeout-in-sec": 150
+    "advance-timeout-in-sec": 150,
+    "debezium-disable-schema": false
   },
   "consistent": {
     "level": "none",
@@ -479,7 +481,8 @@ const (
       "file-size": 1024,
       "output-column-id":false
     },
-    "advance-timeout-in-sec": 150
+    "advance-timeout-in-sec": 150,
+    "debezium-disable-schema": false
   },
   "consistent": {
     "level": "none",

diff --git a/pkg/config/replica_config.go b/pkg/config/replica_config.go
@@ -78,6 +78,7 @@ var defaultReplicaConfig = &ReplicaConfig{
 		ContentCompatible:                util.AddressOf(false),
 		TiDBSourceID:                     1,
 		AdvanceTimeoutInSec:              util.AddressOf(DefaultAdvanceTimeoutInSec),
+		DebeziumDisableSchema:            util.AddressOf(false),
 	},
 	Consistent: &ConsistentConfig{
 		Level:                 "none",

diff --git a/pkg/config/replica_config_test.go b/pkg/config/replica_config_test.go
@@ -69,6 +69,7 @@ func TestReplicaConfigMarshal(t *testing.T) {
 	conf.Sink.ContentCompatible = aws.Bool(true)
 	conf.Sink.SafeMode = aws.Bool(true)
 	conf.Sink.AdvanceTimeoutInSec = util.AddressOf(uint(150))
+	conf.Sink.DebeziumDisableSchema = util.AddressOf(false)
 	conf.Sink.KafkaConfig = &KafkaConfig{
 		PartitionNum:                 aws.Int32(1),
 		ReplicationFactor:            aws.Int16(1),

diff --git a/pkg/orchestrator/reactor_state_test.go b/pkg/orchestrator/reactor_state_test.go
@@ -127,6 +127,7 @@ func TestChangefeedStateUpdate(t *testing.T) {
 							OnlyOutputUpdatedColumns:         config.GetDefaultReplicaConfig().Sink.OnlyOutputUpdatedColumns,
 							DeleteOnlyOutputHandleKeyColumns: config.GetDefaultReplicaConfig().Sink.DeleteOnlyOutputHandleKeyColumns,
 							ContentCompatible:                config.GetDefaultReplicaConfig().Sink.ContentCompatible,
+							DebeziumDisableSchema:            config.GetDefaultReplicaConfig().Sink.DebeziumDisableSchema,
 						},
 						Consistent: config.GetDefaultReplicaConfig().Consistent,
 						Integrity:  config.GetDefaultReplicaConfig().Integrity,
@@ -191,6 +192,7 @@ func TestChangefeedStateUpdate(t *testing.T) {
 							OnlyOutputUpdatedColumns:         config.GetDefaultReplicaConfig().Sink.OnlyOutputUpdatedColumns,
 							DeleteOnlyOutputHandleKeyColumns: config.GetDefaultReplicaConfig().Sink.DeleteOnlyOutputHandleKeyColumns,
 							ContentCompatible:                config.GetDefaultReplicaConfig().Sink.ContentCompatible,
+							DebeziumDisableSchema:            config.GetDefaultReplicaConfig().Sink.DebeziumDisableSchema,
 						},
 						Scheduler:  config.GetDefaultReplicaConfig().Scheduler,
 						Integrity:  config.GetDefaultReplicaConfig().Integrity,
@@ -261,6 +263,7 @@ func TestChangefeedStateUpdate(t *testing.T) {
 							OnlyOutputUpdatedColumns:         config.GetDefaultReplicaConfig().Sink.OnlyOutputUpdatedColumns,
 							DeleteOnlyOutputHandleKeyColumns: config.GetDefaultReplicaConfig().Sink.DeleteOnlyOutputHandleKeyColumns,
 							ContentCompatible:                config.GetDefaultReplicaConfig().Sink.ContentCompatible,
+							DebeziumDisableSchema:            config.GetDefaultReplicaConfig().Sink.DebeziumDisableSchema,
 						},
 						Consistent: config.GetDefaultReplicaConfig().Consistent,
 						Scheduler:  config.GetDefaultReplicaConfig().Scheduler,

diff --git a/tests/integration_tests/api_v2/cases.go b/tests/integration_tests/api_v2/cases.go
@@ -80,6 +80,7 @@ var customReplicaConfig = &ReplicaConfig{
 		EncoderConcurrency:       util.AddressOf(32),
 		EnablePartitionSeparator: util.AddressOf(true),
 		ContentCompatible:        util.AddressOf(true),
+		DebeziumDisableSchema:    util.AddressOf(true),
 	},
 	Scheduler: &ChangefeedSchedulerConfig{
 		EnableTableAcrossNodes: false,
@@ -127,6 +128,7 @@ var defaultReplicaConfig = &ReplicaConfig{
 		EncoderConcurrency:       util.AddressOf(32),
 		EnablePartitionSeparator: util.AddressOf(true),
 		ContentCompatible:        util.AddressOf(false),
+		DebeziumDisableSchema:    util.AddressOf(false),
 	},
 	Scheduler: &ChangefeedSchedulerConfig{
 		EnableTableAcrossNodes: false,

diff --git a/tests/integration_tests/api_v2/model.go b/tests/integration_tests/api_v2/model.go
@@ -244,6 +244,7 @@ type SinkConfig struct {
 	DateSeparator            string            `json:"date_separator,omitempty"`
 	EnablePartitionSeparator *bool             `json:"enable_partition_separator,omitempty"`
 	ContentCompatible        *bool             `json:"content_compatible"`
+	DebeziumDisableSchema    *bool             `json:"debezium_disable_schema,omitempty"`
 }
 
 // CSVConfig denotes the csv config