Merge pull request #8005 from planetscale/schema-fetch

Detect and signal schema changes on vttablets

go/mysql/endtoend/schema_change_test.go

@@ -0,0 +1,114 @@
+/*
+Copyright 2021 The Vitess Authors.
+
+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,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package endtoend
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"vitess.io/vitess/go/mysql"
+)
+
+var ctx = context.Background()
+
+const (
+	createDb        = `create database if not exists _vt`
+	createUserTable = `create table vttest.product (id bigint(20) primary key, name char(10) CHARACTER SET utf8 COLLATE utf8_unicode_ci, created bigint(20))`
+	dropTestTable   = `drop table if exists product`
+)
+
+func TestChangeSchemaIsNoticed(t *testing.T) {
+	conn, err := mysql.Connect(ctx, &connParams)
+	require.NoError(t, err)
+	defer conn.Close()
+
+	_, err = conn.ExecuteFetch(createDb, 1000, true)
+	require.NoError(t, err)
+	_, err = conn.ExecuteFetch(mysql.CreateSchemaCopyTable, 1000, true)
+	require.NoError(t, err)
+
+	tests := []struct {
+		name    string
+		changeQ string
+	}{{
+		name:    "add column",
+		changeQ: "alter table vttest.product add column phone VARCHAR(15)",
+	}, {
+		name:    "rename column",
+		changeQ: "alter table vttest.product change name firstname char(10)",
+	}, {
+		name:    "change column type",
+		changeQ: "alter table vttest.product change name name char(100)",
+	}, {
+		name:    "remove column",
+		changeQ: "alter table vttest.product drop column name",
+	}, {
+		name:    "remove last column",
+		changeQ: "alter table vttest.product drop column created",
+	}, {
+		name:    "remove table",
+		changeQ: "drop table product",
+	}, {
+		name:    "create table",
+		changeQ: `create table vttest.new_table (id bigint(20) primary key)`,
+	}, {
+		name:    "change character set",
+		changeQ: "alter table vttest.product change name name char(10) CHARACTER SET utf8mb4",
+	}, {
+		name:    "change collation",
+		changeQ: "alter table vttest.product change name name char(10) COLLATE utf8_unicode_520_ci",
+	}, {
+		name:    "drop PK",
+		changeQ: "alter table vttest.product drop primary key",
+	}, {
+		name:    "change PK",
+		changeQ: "alter table vttest.product drop primary key, add primary key (name)",
+	}}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			// reset schemacopy
+			_, err := conn.ExecuteFetch(mysql.ClearSchemaCopy, 1000, true)
+			require.NoError(t, err)
+			_, err = conn.ExecuteFetch(dropTestTable, 1000, true)
+			require.NoError(t, err)
+			_, err = conn.ExecuteFetch(createUserTable, 1000, true)
+			require.NoError(t, err)
+			rs, err := conn.ExecuteFetch(mysql.InsertIntoSchemaCopy, 1000, true)
+			require.NoError(t, err)
+			require.NotZero(t, rs.RowsAffected)
+
+			// make sure no changes are detected
+			rs, err = conn.ExecuteFetch(mysql.DetectSchemaChange, 1000, true)
+			require.NoError(t, err)
+			require.Empty(t, rs.Rows)
+
+			// make the schema change
+			_, err = conn.ExecuteFetch(test.changeQ, 1000, true)
+			require.NoError(t, err)
+
+			// make sure the change is detected
+			rs, err = conn.ExecuteFetch(mysql.DetectSchemaChange, 1000, true)
+			require.NoError(t, err)
+			require.NotEmpty(t, rs.Rows)
+		})
+
+	}
+
+}

go/mysql/schema.go

@@ -34,8 +34,74 @@ const (
 	BaseShowPrimary = "SELECT table_name, column_name FROM information_schema.key_column_usage WHERE table_schema=database() AND constraint_name='PRIMARY' ORDER BY table_name, ordinal_position"
 	// ShowRowsRead is the query used to find the number of rows read.
 	ShowRowsRead = "show status like 'Innodb_rows_read'"
+
+	// CreateVTDatabase creates the _vt database
+	CreateVTDatabase = `CREATE DATABASE IF NOT EXISTS _vt`
+
+	// CreateSchemaCopyTable query creates schemacopy table in _vt schema.
+	CreateSchemaCopyTable = `
+CREATE TABLE if not exists _vt.schemacopy (
+	table_schema varchar(64) NOT NULL,
+	table_name varchar(64) NOT NULL,
+	column_name varchar(64) NOT NULL,
+	ordinal_position bigint(21) unsigned NOT NULL,
+	character_set_name varchar(32) DEFAULT NULL,
+	collation_name varchar(32) DEFAULT NULL,
+	column_type longtext NOT NULL,
+	column_key varchar(3) NOT NULL,
+	PRIMARY KEY (table_schema, table_name, ordinal_position))`
+
+	detectNewColumns = `
+select ISC.table_name
+from information_schema.columns as ISC
+	 left join _vt.schemacopy as c on 
+		ISC.table_name = c.table_name and 
+		ISC.table_schema=c.table_schema and 
+		ISC.ordinal_position = c.ordinal_position
+where ISC.table_schema = database() AND c.table_schema is null`
+
+	detectChangeColumns = `
+select ISC.table_name
+from information_schema.columns as ISC
+	  join _vt.schemacopy as c on 
+		ISC.table_name = c.table_name and 
+		ISC.table_schema=c.table_schema and 
+		ISC.ordinal_position = c.ordinal_position
+where ISC.table_schema = database() 
+	AND (not(c.column_name <=> ISC.column_name) 
+	OR not(ISC.character_set_name <=> c.character_set_name) 
+	OR not(ISC.collation_name <=> c.collation_name) 
+	OR not(ISC.column_type <=> c.column_type) 
+	OR not(ISC.column_key <=> c.column_key))`
+
+	detectRemoveColumns = `
+select c.table_name
+from information_schema.columns as ISC
+	  right join _vt.schemacopy as c on 
+		ISC.table_name = c.table_name and 
+		ISC.table_schema=c.table_schema and 
+		ISC.ordinal_position = c.ordinal_position
+where c.table_schema = database() AND ISC.table_schema is null`
+
+	// DetectSchemaChange query detects if there is any schema change from previous copy.
+	DetectSchemaChange = detectChangeColumns + " UNION " + detectNewColumns + " UNION " + detectRemoveColumns
+
+	// ClearSchemaCopy query clears the schemacopy table.
+	ClearSchemaCopy = `delete from _vt.schemacopy`
+
+	// InsertIntoSchemaCopy query copies over the schema information from information_schema.columns table.
+	InsertIntoSchemaCopy = `insert _vt.schemacopy 
+select table_schema, table_name, column_name, ordinal_position, character_set_name, collation_name, column_type, column_key 
+from information_schema.columns 
+where table_schema = database()`
 )
 
+// VTDatabaseInit contains all the schema creation queries needed to
+var VTDatabaseInit = []string{
+	CreateVTDatabase,
+	CreateSchemaCopyTable,
+}
+
 // BaseShowTablesFields contains the fields returned by a BaseShowTables or a BaseShowTablesForTable command.
 // They are validated by the
 // testBaseShowTables test.