sink(ticdc): fix incorrect default value for the not null timestamp f…

…ield in the Debezium protocol (#11976) (#11991)

close #11966

pkg/sink/codec/debezium/helper.go

@@ -0,0 +1,58 @@
+// 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 debezium
+
+import (
+	"encoding/binary"
+	"fmt"
+
+	"github.com/pingcap/tiflow/cdc/model"
+	"github.com/pingcap/tiflow/pkg/sink/codec/common"
+)
+
+func parseBit(s string, n int) string {
+	var result string
+	if len(s) > 0 {
+		// Leading zeros may be omitted
+		result = fmt.Sprintf("%0*b", n%8, s[0])
+	}
+	for i := 1; i < len(s); i++ {
+		result += fmt.Sprintf("%08b", s[i])
+	}
+	return result
+}
+
+func getBitFromUint64(n int, v uint64) []byte {
+	var buf [8]byte
+	binary.LittleEndian.PutUint64(buf[:], v)
+	numBytes := n / 8
+	if n%8 != 0 {
+		numBytes += 1
+	}
+	return buf[:numBytes]
+}
+
+func getValue(col model.ColumnDataX) any {
+	if col.Value == nil {
+		return col.GetDefaultValue()
+	}
+	return col.Value
+}
+
+func getSchemaTopicName(namespace string, schema string, table string) string {
+	return fmt.Sprintf("%s.%s.%s",
+		common.SanitizeName(namespace),
+		common.SanitizeName(schema),
+		common.SanitizeTopicName(table))
+}

pkg/sink/codec/debezium/helper_test.go

@@ -0,0 +1,41 @@
+// 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 debezium
+
+import (
+	"testing"
+
+	"github.com/pingcap/tiflow/cdc/model"
+	"github.com/stretchr/testify/require"
+)
+
+func TestGetValue(t *testing.T) {
+	column := &model.Column{
+		Default: 1,
+	}
+	data := model.Column2ColumnDataXForTest(column)
+	v := getValue(data)
+	require.Equal(t, v, int64(1))
+	data.Value = 2
+	v = getValue(data)
+	require.Equal(t, v, 2)
+}
+
+func TestGetSchemaTopicName(t *testing.T) {
+	namespace := "default"
+	schema := "1A.B"
+	table := "columnNameWith中文"
+	name := getSchemaTopicName(namespace, schema, table)
+	require.Equal(t, name, "default._1A_B.columnNameWith__")
+}

tests/integration_tests/debezium/sql/data_types.sql

@@ -37,8 +37,7 @@ CREATE TABLE t_datetime(
   col_5   DATETIME(5),
   col_6   DATETIME(6),
   col_z   DATETIME DEFAULT 0,
-  -- TODO:
-  -- col_default_current_timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
+  col_default_current_timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
   pk      INT PRIMARY KEY
 );
 
@@ -52,7 +51,7 @@ INSERT INTO t_datetime VALUES (
   '2023-11-16 12:34:56.123456',
   '2023-11-16 12:34:56.123456',
   NULL,
-  -- '2023-11-16 12:34:56.123456',
+  '2023-11-16 12:34:56.123456',
   1
 );
 
@@ -66,7 +65,7 @@ INSERT INTO t_datetime VALUES (
   '2023-11-16 12:34:56',
   '2023-11-16 12:34:56',
   NULL,
-  -- '2023-11-16 12:34:56',
+  '2023-11-16 12:34:56',
   2
 );
 
@@ -82,7 +81,7 @@ INSERT INTO t_datetime VALUES (
   '2023-11-16 12:34:56.123456',
   '2023-11-16 12:34:56.123456',
   NULL,
-  -- '2023-11-16 12:34:56.123456',
+  '2023-11-16 12:34:56.123456',
   3
 );
 
@@ -98,7 +97,7 @@ INSERT INTO t_datetime VALUES (
   NULL,
   NULL,
   NULL,
-  -- NULL,
+  NULL,
   4
 );
 

tests/integration_tests/debezium/sql/debezium/default_value.sql

@@ -0,0 +1,155 @@
+CREATE TABLE UNSIGNED_TINYINT_TABLE (
+  id int PRIMARY KEY,
+  A TINYINT UNSIGNED NULL DEFAULT 0,
+  B TINYINT UNSIGNED NULL DEFAULT '10',
+  C TINYINT UNSIGNED NULL,
+  D TINYINT UNSIGNED NOT NULL,
+  E TINYINT UNSIGNED NOT NULL DEFAULT 0,
+  F TINYINT UNSIGNED NOT NULL DEFAULT '0',
+  G TINYINT UNSIGNED NULL DEFAULT '100'
+);
+INSERT INTO UNSIGNED_TINYINT_TABLE VALUES (1, DEFAULT, DEFAULT, 0, 1, DEFAULT, DEFAULT, NULL);
+
+CREATE TABLE UNSIGNED_SMALLINT_TABLE (
+  id int PRIMARY KEY,
+  A SMALLINT UNSIGNED NULL DEFAULT 0,
+  B SMALLINT UNSIGNED NULL DEFAULT '10',
+  C SMALLINT UNSIGNED NULL,
+  D SMALLINT UNSIGNED NOT NULL,
+  E SMALLINT UNSIGNED NOT NULL DEFAULT 0,
+  F SMALLINT UNSIGNED NOT NULL DEFAULT '0',
+  G SMALLINT UNSIGNED NULL DEFAULT '100'
+);
+INSERT INTO UNSIGNED_SMALLINT_TABLE VALUES (1, 1, 1, 1, 0, 1, 1, NULL);
+
+CREATE TABLE UNSIGNED_MEDIUMINT_TABLE (
+  id int PRIMARY KEY,
+  A MEDIUMINT UNSIGNED NULL DEFAULT 0,
+  B MEDIUMINT UNSIGNED NULL DEFAULT '10',
+  C MEDIUMINT UNSIGNED NULL,
+  D MEDIUMINT UNSIGNED NOT NULL,
+  E MEDIUMINT UNSIGNED NOT NULL DEFAULT 0,
+  F MEDIUMINT UNSIGNED NOT NULL DEFAULT '0',
+  G MEDIUMINT UNSIGNED NULL DEFAULT '100'
+);
+INSERT INTO UNSIGNED_MEDIUMINT_TABLE VALUES (1, 1, 1, 1, 0, 1, 1, NULL);
+
+CREATE TABLE UNSIGNED_INT_TABLE (
+  id int PRIMARY KEY,
+  A INT UNSIGNED NULL DEFAULT 0,
+  B INT UNSIGNED NULL DEFAULT '10',
+  C INT UNSIGNED NULL,
+  D INT UNSIGNED NOT NULL,
+  E INT UNSIGNED NOT NULL DEFAULT 0,
+  F INT UNSIGNED NOT NULL DEFAULT '0',
+  G INT UNSIGNED NULL DEFAULT '100'
+);
+INSERT INTO UNSIGNED_INT_TABLE VALUES (1, 1, 1, 1, 0, 1, 1, NULL);
+
+CREATE TABLE UNSIGNED_BIGINT_TABLE (
+  id int PRIMARY KEY,
+  A BIGINT UNSIGNED NULL DEFAULT 0,
+  B BIGINT UNSIGNED NULL DEFAULT '10',
+  C BIGINT UNSIGNED NULL,
+  D BIGINT UNSIGNED NOT NULL,
+  E BIGINT UNSIGNED NOT NULL DEFAULT 0,
+  F BIGINT UNSIGNED NOT NULL DEFAULT '0',
+  G BIGINT UNSIGNED NULL DEFAULT '100'
+);
+INSERT INTO UNSIGNED_BIGINT_TABLE VALUES (1, 1, 1, 1, 0, 1, 1, NULL);
+
+CREATE TABLE STRING_TABLE (
+  id int PRIMARY KEY,
+  A CHAR(1) NULL DEFAULT 'A',
+  B CHAR(1) NULL DEFAULT 'b',
+  C VARCHAR(10) NULL DEFAULT 'CC',
+  F CHAR(1) DEFAULT NULL,
+  G VARCHAR(10) DEFAULT NULL,
+  I VARCHAR(10) NULL DEFAULT '100'
+);
+INSERT INTO STRING_TABLE
+VALUES (1, DEFAULT, DEFAULT, DEFAULT, DEFAULT, DEFAULT, NULL);
+
+CREATE TABLE BIT_TABLE (
+  id int PRIMARY KEY,
+  A BIT(1) NULL DEFAULT NULL,
+  B BIT(1) DEFAULT 0,
+  C BIT(1) DEFAULT 1,
+  D BIT(1) DEFAULT b'0',
+  E BIT(1) DEFAULT b'1',
+  H BIT(10) DEFAULT b'0101000010',
+  I BIT(10) DEFAULT NULL,
+  J BIT(25) DEFAULT b'0000000000000000100001111',
+  K BIT(25) DEFAULT b'0000000010110000100001111'
+);
+INSERT INTO BIT_TABLE
+VALUES (1, false, DEFAULT, DEFAULT, DEFAULT, DEFAULT, DEFAULT, NULL, DEFAULT, NULL);
+
+CREATE TABLE NUMBER_TABLE (
+  id int PRIMARY KEY,
+  A TINYINT NULL DEFAULT 10,
+  B SMALLINT NOT NULL DEFAULT '5',
+  C INTEGER NOT NULL DEFAULT 0,
+  D BIGINT NOT NULL DEFAULT 20,
+  E INT NULL DEFAULT NULL,
+  F INT NULL DEFAULT 30
+);
+INSERT INTO NUMBER_TABLE
+VALUES (1, DEFAULT, DEFAULT, DEFAULT, DEFAULT, DEFAULT, NULL);
+
+CREATE TABLE FlOAT_DOUBLE_TABLE (
+  id int PRIMARY KEY,
+  F FLOAT NULL DEFAULT 0,
+  G DOUBLE NOT NULL DEFAULT 1.1,
+  H DOUBLE NULL DEFAULT 3.3
+);
+INSERT INTO FlOAT_DOUBLE_TABLE
+VALUES (1, DEFAULT, DEFAULT, NULL);
+
+-- set sql_mode REAL_AS_FLOAT is necessary
+set @@session.sql_mode=concat(@@session.sql_mode, ',REAL_AS_FLOAT');
+CREATE TABLE REAL_TABLE (
+  id int PRIMARY KEY,
+  A REAL NOT NULL DEFAULT 1,
+  B REAL NULL DEFAULT NULL,
+  C REAL NULL DEFAULT 3
+);
+INSERT INTO REAL_TABLE
+VALUES (1, DEFAULT ,DEFAULT, NULL);
+
+CREATE TABLE NUMERIC_DECIMAL_TABLE (
+  id int PRIMARY KEY,
+  A NUMERIC(3, 2) NOT NULL DEFAULT 1.23,
+  B DECIMAL(4, 3) NOT NULL DEFAULT 2.321,
+  C NUMERIC(7, 5) NULL DEFAULT '12.678',
+  D NUMERIC(7, 5) NULL DEFAULT '15.28'
+);
+INSERT INTO NUMERIC_DECIMAL_TABLE
+VALUES (1, 1.33 ,2.111 , 3.444, NULL);
+
+CREATE TABLE DATE_TIME_TABLE (
+  id int PRIMARY KEY,
+  A DATE NOT NULL DEFAULT '1976-08-23',
+  C DATETIME DEFAULT '2018-01-03 00:00:10',
+  D DATETIME(1) DEFAULT '2018-01-03 00:00:10.7',
+  E DATETIME(6) DEFAULT '2018-01-03 00:00:10.123456',
+  F YEAR NOT NULL DEFAULT 1,
+  G TIME DEFAULT '00:00:00',
+  H TIME(1) DEFAULT '23:00:00.7',
+  I TIME(6) DEFAULT '23:00:00.123456',
+  J TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  K TIMESTAMP NULL DEFAULT CURRENT_TIMESTAMP,
+  L TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  M TIME(1) DEFAULT '-23:45:56.7',
+  N TIME(6) DEFAULT '123:00:00.123456'
+);
+INSERT INTO DATE_TIME_TABLE
+VALUES (1, DEFAULT, DEFAULT, DEFAULT, DEFAULT, DEFAULT, DEFAULT, DEFAULT, DEFAULT, DEFAULT, NULL, DEFAULT, DEFAULT, DEFAULT);
+
+CREATE TABLE DBZ_771_CUSTOMERS (
+  id INTEGER NOT NULL PRIMARY KEY,
+  CUSTOMER_TYPE ENUM ('b2c','b2b') NOT NULL default 'b2c'
+);
+
+INSERT INTO DBZ_771_CUSTOMERS
+VALUES (1, 'b2b');

tests/integration_tests/debezium/sql/debezium/default_value_all_zero_time.sql

@@ -0,0 +1,33 @@
+SET sql_mode='';
+CREATE TABLE all_zero_date_and_time_table (
+  id INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
+  A TIMESTAMP NOT NULL DEFAULT '0000-00-00 00:00:00',
+  B TIMESTAMP NULL DEFAULT '0000-00-00 00:00:00',
+  C TIMESTAMP DEFAULT '0000-00-00 00:00:00' NOT NULL,
+  D TIMESTAMP DEFAULT '0000-00-00 00:00:00' NULL,
+  E DATE NOT NULL DEFAULT '0000-00-00',
+  F DATE NULL DEFAULT '0000-00-00',
+  G DATE DEFAULT '0000-00-00' NOT NULL,
+  H DATE DEFAULT '0000-00-00' NULL,
+  I DATETIME NOT NULL DEFAULT '0000-00-00 00:00:00',
+  J DATETIME NULL DEFAULT '0000-00-00 00:00:00',
+  K DATETIME DEFAULT '0000-00-00 00:00:00' NOT NULL,
+  L DATETIME DEFAULT '0000-00-00 00:00:00' NULL
+);
+INSERT INTO all_zero_date_and_time_table
+VALUES (DEFAULT, DEFAULT, DEFAULT, DEFAULT, DEFAULT, DEFAULT, DEFAULT, DEFAULT, DEFAULT, DEFAULT, DEFAULT, DEFAULT, DEFAULT);
+
+CREATE TABLE part_zero_date_and_time_table (
+  id INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
+  A TIMESTAMP NOT NULL DEFAULT '0000-00-00 00:00:00',
+  B TIMESTAMP NULL DEFAULT '0000-00-00 00:00:00',
+  C DATETIME NOT NULL DEFAULT '0000-00-00 01:00:00.000',
+  D DATETIME NULL DEFAULT '0000-00-00 01:00:00.000',
+  E DATE NOT NULL DEFAULT '1000-00-00',
+  /* TiCDC convert '1000-00-00 01:00:00.000' to "1000-00-00"
+  F DATE NULL DEFAULT '1000-00-00 01:00:00.000',*/
+  G TIME DEFAULT '0000-00-00 01:00:00.000' NOT NULL,
+  H TIME DEFAULT '0000-00-00 01:00:00.000' NULL
+);
+INSERT INTO part_zero_date_and_time_table
+VALUES (DEFAULT, DEFAULT, DEFAULT, DEFAULT, DEFAULT, /*DEFAULT,*/ DEFAULT, DEFAULT, DEFAULT);

tests/integration_tests/debezium/sql/debezium/default_value_generated.sql

@@ -0,0 +1,8 @@
+/* see https://github.com/pingcap/tiflow/issues/11704 */
+CREATE TABLE GENERATED_TABLE (
+  id int PRIMARY KEY,
+  A SMALLINT UNSIGNED,
+  B SMALLINT UNSIGNED AS (2 * A) STORED,
+  C SMALLINT UNSIGNED AS (3 * A) STORED NOT NULL
+);
+INSERT INTO GENERATED_TABLE VALUES (1, 15, DEFAULT, DEFAULT);

tests/integration_tests/debezium/sql/debezium/year_test.sql

@@ -0,0 +1,45 @@
+SET time_zone='UTC';
+CREATE TABLE dbz_1143_year_test (
+      id INT AUTO_INCREMENT NOT NULL,
+      y18 YEAR,
+      y0018 YEAR,
+      y2018 YEAR,
+      d18 DATE,
+      d0018 DATE,
+      d2018 DATE,
+      dt18 DATETIME,
+      dt0018 DATETIME,
+      dt2018 DATETIME,
+      y78 YEAR,
+      y0078 YEAR,
+      y1978 YEAR,
+      d78 DATE,
+      d0078 DATE,
+      d1978 DATE,
+      dt78 DATETIME,
+      dt0078 DATETIME,
+      dt1978 DATETIME,
+     PRIMARY KEY (id)
+) DEFAULT CHARSET=utf8;
+
+INSERT INTO dbz_1143_year_test VALUES (
+    default,
+    '18',
+    '0018',
+    '2018',
+    '18-04-01',
+    '0018-04-01',
+    '2018-04-01',
+    '18-04-01 12:34:56',
+    '0018-04-01 12:34:56',
+    '2018-04-01 12:34:56',
+    '78',
+    '0078',
+    '1978',
+    '78-04-01',
+    '0078-04-01',
+    '1978-04-01',
+    '78-04-01 12:34:56',
+    '0078-04-01 12:34:56',
+    '1978-04-01 12:34:56'
+);