executor: filter on table_schema for INFORMATION_SCHEMA.KEY_COLUMN_USAGE table (#55162)

close #55156

Author: YangKeao

pkg/executor/infoschema_reader.go

@@ -1842,9 +1842,13 @@ func (e *memtableRetriever) setDataFromKeyColumnUsage(ctx context.Context, sctx
 		return nil
 	}
 	for _, schema := range schemas {
+		// `constraint_schema` and `table_schema` are always the same in MySQL.
 		if ok && extractor.Filter("constraint_schema", schema.L) {
 			continue
 		}
+		if ok && extractor.Filter("table_schema", schema.L) {
+			continue
+		}
 		tables, err := e.is.SchemaTableInfos(ctx, schema)
 		if err != nil {
 			return errors.Trace(err)

tests/integrationtest/r/infoschema/infoschema.result

@@ -159,3 +159,15 @@ pt2	p2
 pt2	p3
 select TABLE_NAME, PARTITION_NAME from information_schema.partitions where table_name = 'pt0' and table_schema = 'infoschema__infoschema';
 TABLE_NAME	PARTITION_NAME
+create database if not exists db1;
+create table db1.table1(id int not null primary key, cat_name varchar(255) not null, cat_description text);
+create table db1.table2(id int not null, FOREIGN KEY fk(id) REFERENCES table1(id) ON UPDATE CASCADE ON DELETE RESTRICT);
+create database if not exists db2;
+create table db2.table1(id int not null primary key, cat_name varchar(255) not null, cat_description text);
+create table db2.table2(id int not null, FOREIGN KEY fk(id) REFERENCES table1(id) ON UPDATE CASCADE ON DELETE RESTRICT);
+select * from INFORMATION_SCHEMA.KEY_COLUMN_USAGE where table_schema = 'db1' order by TABLE_NAME;
+CONSTRAINT_CATALOG	CONSTRAINT_SCHEMA	CONSTRAINT_NAME	TABLE_CATALOG	TABLE_SCHEMA	TABLE_NAME	COLUMN_NAME	ORDINAL_POSITION	POSITION_IN_UNIQUE_CONSTRAINT	REFERENCED_TABLE_SCHEMA	REFERENCED_TABLE_NAME	REFERENCED_COLUMN_NAME
+def	db1	PRIMARY	def	db1	table1	id	1	1	NULL	NULL	NULL
+def	db1	fk	def	db1	table2	id	1	1	db1	table1	id
+drop database db1;
+drop database db2;

tests/integrationtest/t/infoschema/infoschema.test

@@ -69,3 +69,14 @@ select TABLE_NAME, PARTITION_NAME from information_schema.partitions where table
 select TABLE_NAME, PARTITION_NAME from information_schema.partitions where table_name = 'pt2' and table_schema = 'infoschema__infoschema';
 -- sorted_result
 select TABLE_NAME, PARTITION_NAME from information_schema.partitions where table_name = 'pt0' and table_schema = 'infoschema__infoschema';
+
+# TestFilterKeyColumnUsageTable
+create database if not exists db1;
+create table db1.table1(id int not null primary key, cat_name varchar(255) not null, cat_description text);
+create table db1.table2(id int not null, FOREIGN KEY fk(id) REFERENCES table1(id) ON UPDATE CASCADE ON DELETE RESTRICT);
+create database if not exists db2;
+create table db2.table1(id int not null primary key, cat_name varchar(255) not null, cat_description text);
+create table db2.table2(id int not null, FOREIGN KEY fk(id) REFERENCES table1(id) ON UPDATE CASCADE ON DELETE RESTRICT);
+select * from INFORMATION_SCHEMA.KEY_COLUMN_USAGE where table_schema = 'db1' order by TABLE_NAME;
+drop database db1;
+drop database db2;