Add VisibilityBitmap to TableMapEvent in replication (#813)

* Add `VisibilityBitmap` to `TableMapEvent` for MySQL 8.0.23+ Invisible Columns

Changes:
- `TableMapEvent.VisibilityBitmap`
`VisibilityBitmap` is a bitmap where each bit represents the visibility of a corresponding column in a table.
If a bit is set, it indicates that the corresponding column is NOT an invinsible column.
Invisible column was introduced in MySQL version 8.0.23.

- `TableMapEvent.VisibilityMap`
`VisibilityMap` lists boolean values of which is true if column of the same index is NOT an invisible column.

Co-authored-by: sean <[email protected]>

* Refactor bitmap iteration to align with MySQL source code

Changes:
- Refactpred `UnsignedMap`
- Refactored `VisibilityMap`

Suggested by: #813 (comment)

* Add test for column visibility in table map

Changes:
- Added data for MySQL 8.0 only (Only MySQL 8.0.23+ supports invisible columns)

* Add test case 2 for column visibility in table map

Changes:
- Add test case 2 where invisible columns does not exists at all

---------

Co-authored-by: sean <[email protected]>

Author: dongwook-chan

replication/const.go

@@ -240,6 +240,7 @@ const (
 	TABLE_MAP_OPT_META_PRIMARY_KEY_WITH_PREFIX
 	TABLE_MAP_OPT_META_ENUM_AND_SET_DEFAULT_CHARSET
 	TABLE_MAP_OPT_META_ENUM_AND_SET_COLUMN_CHARSET
+	TABLE_MAP_OPT_META_COLUMN_VISIBILITY
 )
 
 type IntVarEventType byte

replication/row_event.go

@@ -81,6 +81,9 @@ type TableMapEvent struct {
 	// EnumSetDefaultCharset/EnumSetColumnCharset is similar to DefaultCharset/ColumnCharset but for enum/set columns.
 	EnumSetDefaultCharset []uint64
 	EnumSetColumnCharset  []uint64
+
+	// VisibilityBitmap stores bits that are set if corresponding column is not invisible (MySQL 8.0.23+)
+	VisibilityBitmap []byte
 }
 
 func (e *TableMapEvent) Decode(data []byte) error {
@@ -312,6 +315,9 @@ func (e *TableMapEvent) decodeOptionalMeta(data []byte) (err error) {
 				return err
 			}
 
+		case TABLE_MAP_OPT_META_COLUMN_VISIBILITY:
+			e.VisibilityBitmap = v
+
 		default:
 			// Ignore for future extension
 		}
@@ -421,6 +427,7 @@ func (e *TableMapEvent) Dump(w io.Writer) {
 	fmt.Fprintf(w, "Primary key prefix: %v\n", e.PrimaryKeyPrefix)
 	fmt.Fprintf(w, "Enum/set default charset: %v\n", e.EnumSetDefaultCharset)
 	fmt.Fprintf(w, "Enum/set column charset: %v\n", e.EnumSetColumnCharset)
+	fmt.Fprintf(w, "Invisible Column bitmap: \n%s", hex.Dump(e.VisibilityBitmap))
 
 	unsignedMap := e.UnsignedMap()
 	fmt.Fprintf(w, "UnsignedMap: %#v\n", unsignedMap)
@@ -440,6 +447,9 @@ func (e *TableMapEvent) Dump(w io.Writer) {
 	geometryTypeMap := e.GeometryTypeMap()
 	fmt.Fprintf(w, "GeometryTypeMap: %#v\n", geometryTypeMap)
 
+	visibilityMap := e.VisibilityMap()
+	fmt.Fprintf(w, "VisibilityMap: %#v\n", visibilityMap)
+
 	nameMaxLen := 0
 	for _, name := range e.ColumnName {
 		if len(name) > nameMaxLen {
@@ -608,14 +618,19 @@ func (e *TableMapEvent) UnsignedMap() map[int]bool {
 	if len(e.SignednessBitmap) == 0 {
 		return nil
 	}
-	p := 0
 	ret := make(map[int]bool)
-	for i := 0; i < int(e.ColumnCount); i++ {
-		if !e.IsNumericColumn(i) {
-			continue
+	i := 0
+	for _, field := range e.SignednessBitmap {
+		for c := 0x80; c != 0; {
+			if e.IsNumericColumn(i) {
+				ret[i] = field&byte(c) != 0
+				c >>= 1
+			}
+			i++
+			if i >= int(e.ColumnCount) {
+				return ret
+			}
 		}
-		ret[i] = e.SignednessBitmap[p/8]&(1<<uint(7-p%8)) != 0
-		p++
 	}
 	return ret
 }
@@ -730,6 +745,27 @@ func (e *TableMapEvent) GeometryTypeMap() map[int]uint64 {
 	return ret
 }
 
+// VisibilityMap returns a map: column index -> visiblity.
+// Invisible column was introduced in MySQL 8.0.23
+// nil is returned if not available.
+func (e *TableMapEvent) VisibilityMap() map[int]bool {
+	if len(e.VisibilityBitmap) == 0 {
+		return nil
+	}
+	ret := make(map[int]bool)
+	i := 0
+	for _, field := range e.VisibilityBitmap {
+		for c := 0x80; c != 0; c >>= 1 {
+			ret[i] = field&byte(c) != 0
+			i++
+			if uint64(i) >= e.ColumnCount {
+				return ret
+			}
+		}
+	}
+	return ret
+}
+
 // Below realType and IsXXXColumn are base from:
 //   table_def::type in sql/rpl_utility.h
 //   Table_map_log_event::print_columns in mysql-8.0/sql/log_event.cc and mariadb-10.5/sql/log_event_client.cc

replication/row_event_test.go

@@ -939,6 +939,123 @@ func TestTableMapOptMetaPrimaryKey(t *testing.T) {
 	}
 }
 
+func TestTableMapOptMetaVisibility(t *testing.T) {
+	/*
+		SET GLOBAL binlog_row_image = FULL;
+		SET GLOBAL binlog_row_metadata = FULL;	-- if applicable
+
+		CREATE DATABASE test;
+		USE test;
+	*/
+
+	/*
+		CREATE TABLE _visibility(
+		    `col0` INT INVISIBLE,
+		    `col1` INT,
+		    `col2` INT INVISIBLE,
+		    `col3` INT,
+		    `col4` INT,
+		    `col5` INT INVISIBLE,
+		    `col6` INT INVISIBLE,
+		    `col7` INT INVISIBLE,
+		    `col8` INT,
+		    `col9` INT INVISIBLE,
+		    `col10` INT INVISIBLE
+		);
+	*/
+	case1VisibilityBitmap := []byte{0x58, 0x80}
+	case1VisibilityMap := map[int]bool{
+		0:  false,
+		1:  true,
+		2:  false,
+		3:  true,
+		4:  true,
+		5:  false,
+		6:  false,
+		7:  false,
+		8:  true,
+		9:  false,
+		10: false,
+	}
+
+	/*
+		CREATE TABLE _visibility(
+		    `col0` INT,
+		    `col1` INT,
+		    `col2` INT,
+		    `col3` INT,
+		    `col4` INT,
+		    `col5` INT,
+		    `col6` INT,
+		    `col7` INT,
+		    `col8` INT,
+		    `col9` INT,
+		    `col10` INT
+		);
+	*/
+	case2VisibilityBitmap := []byte{0xff, 0xe0}
+	case2VisibilityMap := map[int]bool{
+		0:  true,
+		1:  true,
+		2:  true,
+		3:  true,
+		4:  true,
+		5:  true,
+		6:  true,
+		7:  true,
+		8:  true,
+		9:  true,
+		10: true,
+	}
+
+	// Invisible column and INVISIBLE keyword is available only on MySQL 8.0.23+
+	testcases := []struct {
+		data                     []byte
+		expectedVisibilityBitmap []byte
+		expectedVisibilityMap    map[int]bool
+	}{
+		{
+			// mysql 8.0, case1
+			data:                     []byte("^\x00\x00\x00\x00\x00\x01\x00\x04test\x00\x0b_visibility\x00\x0b\x03\x03\x03\x03\x03\x03\x03\x03\x03\x03\x03\x00\xff\x07\x01\x02\x00\x00\x048\x04col0\x04col1\x04col2\x04col3\x04col4\x04col5\x04col6\x04col7\x04col8\x04col9\x05col10\x0c\x02X\x80"),
+			expectedVisibilityBitmap: case1VisibilityBitmap,
+			expectedVisibilityMap:    case1VisibilityMap,
+		},
+		{
+			// mysql 5.7, case2
+			data:                     []byte("m\x00\x00\x00\x00\x00\x01\x00\x04test\x00\x0b_visibility\x00\x0b\x03\x03\x03\x03\x03\x03\x03\x03\x03\x03\x03\x00\xff\x07"),
+			expectedVisibilityBitmap: []byte(nil),
+			expectedVisibilityMap:    map[int]bool(nil),
+		},
+		{
+			// mysql 8.0, case2
+			data:                     []byte("^\x00\x00\x00\x00\x00\x01\x00\x04test\x00\x0b_visibility\x00\x0b\x03\x03\x03\x03\x03\x03\x03\x03\x03\x03\x03\x00\xff\x07\x01\x02\x00\x00\x048\x04col0\x04col1\x04col2\x04col3\x04col4\x04col5\x04col6\x04col7\x04col8\x04col9\x05col10\x0c\x02\xff\xe0"),
+			expectedVisibilityBitmap: case2VisibilityBitmap,
+			expectedVisibilityMap:    case2VisibilityMap,
+		},
+		{
+			// mariadb 10.4, case2
+			data:                     []byte("\x12\x00\x00\x00\x00\x00\x01\x00\x04test\x00\x0b_visibility\x00\x0b\x03\x03\x03\x03\x03\x03\x03\x03\x03\x03\x03\x00\xff\x07"),
+			expectedVisibilityBitmap: []byte(nil),
+			expectedVisibilityMap:    map[int]bool(nil),
+		},
+		{
+			// mariadb 10.5, case2
+			data:                     []byte("\x12\x00\x00\x00\x00\x00\x01\x00\x04test\x00\x0b_visibility\x00\x0b\x03\x03\x03\x03\x03\x03\x03\x03\x03\x03\x03\x00\xff\x07\x01\x02\x00\x00\x048\x04col0\x04col1\x04col2\x04col3\x04col4\x04col5\x04col6\x04col7\x04col8\x04col9\x05col10"),
+			expectedVisibilityBitmap: []byte(nil),
+			expectedVisibilityMap:    map[int]bool(nil),
+		},
+	}
+
+	for _, tc := range testcases {
+		tableMapEvent := new(TableMapEvent)
+		tableMapEvent.tableIDSize = 6
+		err := tableMapEvent.Decode(tc.data)
+		require.NoError(t, err)
+		require.Equal(t, tc.expectedVisibilityBitmap, tableMapEvent.VisibilityBitmap)
+		require.Equal(t, tc.expectedVisibilityMap, tableMapEvent.VisibilityMap())
+	}
+}
+
 func TestTableMapHelperMaps(t *testing.T) {
 	/*
 		CREATE TABLE `_types` (