fix: expose table grants in information_schema.TABLE_PRIVILEGES

pkg/bootstrap/versions/v4_0_6/tenant_upgrade_list.go

@@ -47,6 +47,7 @@ var tenantUpgEntries = []versions.UpgradeEntry{
 	backfillUserDefinedFunctionArgumentTypes(),
 	addUserDefinedFunctionSignatureIndex(),
 	upgradeInformationSchemaCollationCharacterSetApplicability(),
+	upgradeInformationSchemaTablePrivileges(),
 }
 
 // Keep this as a separate upgrade entry so tenants that already completed
@@ -160,6 +161,23 @@ func upgradeInformationSchemaCollationCharacterSetApplicability() versions.Upgra
 	}
 }
 
+// upgradeInformationSchemaTablePrivileges replaces the historical empty base
+// table with the canonical catalog-backed view. The three statements converge
+// absent, base-table, and stale-view states without relying on a default schema.
+func upgradeInformationSchemaTablePrivileges() versions.UpgradeEntry {
+	return versions.UpgradeEntry{
+		Schema:    sysview.InformationDBConst,
+		TableName: "TABLE_PRIVILEGES",
+		UpgType:   versions.MODIFY_VIEW,
+		UpgSql: fmt.Sprintf("DROP VIEW IF EXISTS %s.TABLE_PRIVILEGES;",
+			sysview.InformationDBConst),
+		CheckFunc: checkViewDefinition("TABLE_PRIVILEGES", sysview.InformationSchemaTablePrivilegesDDL),
+		PreSql: fmt.Sprintf("DROP TABLE IF EXISTS %s.TABLE_PRIVILEGES;",
+			sysview.InformationDBConst),
+		PostSql: sysview.InformationSchemaTablePrivilegesDDL,
+	}
+}
+
 // User-defined function lookup is scoped by database and argument signature.
 // A global unique name index prevents a database clone from retaining the
 // source and destination function definitions in the same account.

pkg/bootstrap/versions/v4_0_6/upgrade_test.go

@@ -37,7 +37,7 @@ import (
 )
 
 func TestUpgradeEntries(t *testing.T) {
-	require.Len(t, tenantUpgEntries, 19)
+	require.Len(t, tenantUpgEntries, 20)
 	require.Len(t, clusterUpgEntries, 3)
 	require.Equal(t, retireKafkaSinkDaemonTasks.UpgSql, clusterUpgEntries[0].UpgSql)
 	require.Equal(t, catalog.MO_VIEW_DEPENDENCIES, clusterUpgEntries[1].TableName)
@@ -126,6 +126,15 @@ func TestUpgradeEntries(t *testing.T) {
 	require.Equal(t, sysview.InformationSchemaCollationCharacterSetApplicabilityDDL, collationApplicability.UpgSql)
 	require.Contains(t, strings.ToLower(collationApplicability.PreSql),
 		"drop view if exists information_schema.collation_character_set_applicability")
+	tablePrivileges := tenantUpgEntries[19]
+	require.Equal(t, versions.MODIFY_VIEW, tablePrivileges.UpgType)
+	require.Equal(t, sysview.InformationDBConst, tablePrivileges.Schema)
+	require.Equal(t, "TABLE_PRIVILEGES", tablePrivileges.TableName)
+	require.Contains(t, strings.ToLower(tablePrivileges.PreSql),
+		"drop table if exists information_schema.table_privileges")
+	require.Contains(t, strings.ToLower(tablePrivileges.UpgSql),
+		"drop view if exists information_schema.table_privileges")
+	require.Equal(t, sysview.InformationSchemaTablePrivilegesDDL, tablePrivileges.PostSql)
 }
 
 func TestInformationSchemaCollationsUpgradeCheckIsExact(t *testing.T) {
@@ -161,7 +170,7 @@ func TestUserDefinedFunctionArgumentTypesBackfillRejectsOversizedSignature(t *te
 }
 
 func TestForeignKeyMetadataTenantUpgradeEntries(t *testing.T) {
-	require.Len(t, tenantUpgEntries, 19)
+	require.Len(t, tenantUpgEntries, 20)
 
 	for i, column := range []string{"referenced_index_name", "on_delete_origin", "on_update_origin"} {
 		entry := tenantUpgEntries[2+i]
@@ -398,6 +407,7 @@ func TestTenantViewDefinitionChecks(t *testing.T) {
 		upgradeInformationSchemaCheckConstraints(),
 		upgradeInformationSchemaTableConstraints(),
 		upgradeInformationSchemaCollationCharacterSetApplicability(),
+		upgradeInformationSchemaTablePrivileges(),
 	}
 
 	for _, entry := range entries {
@@ -514,6 +524,37 @@ func TestKeyColumnUsageViewUpgradeIsOrderedAndIdempotent(t *testing.T) {
 	require.Empty(t, executed)
 }
 
+func TestTablePrivilegesViewUpgradeIsOrderedAndIdempotent(t *testing.T) {
+	entry := upgradeInformationSchemaTablePrivileges()
+	upgraded := false
+	stub := gostub.Stub(&versions.CheckViewDefinition, func(_ executor.TxnExecutor, accountID uint32, schema, viewName string) (bool, string, error) {
+		require.Equal(t, uint32(42), accountID)
+		require.Equal(t, sysview.InformationDBConst, schema)
+		require.Equal(t, "TABLE_PRIVILEGES", viewName)
+		if upgraded {
+			return true, sysview.InformationSchemaTablePrivilegesDDL, nil
+		}
+		return false, "", nil
+	})
+	defer stub.Reset()
+
+	var executed []string
+	txnExecutor := newVersionTxnExecutor(t, func(sql string) (executor.Result, error) {
+		executed = append(executed, sql)
+		if sql == entry.PostSql {
+			upgraded = true
+		}
+		return executor.Result{}, nil
+	})
+
+	require.NoError(t, entry.Upgrade(txnExecutor, 42))
+	require.Equal(t, []string{entry.PreSql, entry.UpgSql, entry.PostSql}, executed)
+
+	executed = nil
+	require.NoError(t, entry.Upgrade(txnExecutor, 42))
+	require.Empty(t, executed)
+}
+
 func TestVersionHandleLifecycleWithNoLegacyDefinitions(t *testing.T) {
 	runtime.RunTest("", func(runtime.Runtime) {
 		tableStub := gostub.Stub(&versions.CheckTableDefinition, func(executor.TxnExecutor, uint32, string, string) (bool, error) {
@@ -531,6 +572,8 @@ func TestVersionHandleLifecycleWithNoLegacyDefinitions(t *testing.T) {
 				return true, sysview.InformationSchemaCheckConstraintsDDL, nil
 			case "COLLATION_CHARACTER_SET_APPLICABILITY":
 				return true, sysview.InformationSchemaCollationCharacterSetApplicabilityDDL, nil
+			case "TABLE_PRIVILEGES":
+				return true, sysview.InformationSchemaTablePrivilegesDDL, nil
 			case "TABLE_CONSTRAINTS":
 				return true, sysview.InformationSchemaTableConstraintsDDL, nil
 			case "COLUMNS":

pkg/util/sysview/predefined.go

@@ -512,14 +512,33 @@ var (
 		"`IS_GRANTABLE` varchar(3) NOT NULL DEFAULT ''" +
 		")"
 
-	InformationSchemaTablePrivilegesDDL = "CREATE TABLE information_schema.`TABLE_PRIVILEGES` (" +
-		"`GRANTEE` varchar(292) NOT NULL DEFAULT ''," +
-		"`TABLE_CATALOG` varchar(512) NOT NULL DEFAULT ''," +
-		"`TABLE_SCHEMA` varchar(64) NOT NULL DEFAULT ''," +
-		"`TABLE_NAME` varchar(64) NOT NULL DEFAULT ''," +
-		"`PRIVILEGE_TYPE` varchar(64) NOT NULL DEFAULT ''," +
-		"`IS_GRANTABLE` varchar(3) NOT NULL DEFAULT ''" +
-		")"
+	InformationSchemaTablePrivilegesDDL = "CREATE VIEW information_schema.`TABLE_PRIVILEGES` AS " +
+		"SELECT " +
+		"CAST(rp.role_name AS varchar(292)) AS `GRANTEE`," +
+		"CAST('def' AS varchar(512)) AS `TABLE_CATALOG`," +
+		"CAST(tbl.reldatabase AS varchar(64)) AS `TABLE_SCHEMA`," +
+		"CAST(tbl.relname AS varchar(64)) AS `TABLE_NAME`," +
+		"CAST(privilege_map.external_name AS varchar(64)) AS `PRIVILEGE_TYPE`," +
+		"CAST(case when max(case when rp.with_grant_option or rp.privilege_name = 'table ownership' " +
+		"then 1 else 0 end) = 1 then 'YES' else 'NO' end AS varchar(3)) AS `IS_GRANTABLE` " +
+		"FROM mo_catalog.mo_role_privs rp " +
+		"JOIN mo_catalog.mo_tables tbl ON rp.obj_id = tbl.rel_id " +
+		"JOIN (VALUES " +
+		"ROW('select', 'SELECT')," +
+		"ROW('insert', 'INSERT')," +
+		"ROW('update', 'UPDATE')," +
+		"ROW('truncate', 'TRUNCATE')," +
+		"ROW('delete', 'DELETE')," +
+		"ROW('reference', 'REFERENCES')," +
+		"ROW('index', 'INDEX')," +
+		"ROW('values', 'VALUES')" +
+		") privilege_map(internal_name, external_name) " +
+		"ON rp.privilege_name = privilege_map.internal_name " +
+		"OR rp.privilege_name IN ('table all', 'table ownership') " +
+		"WHERE tbl.account_id = current_account_id() " +
+		"AND rp.obj_type IN ('table', 'view') " +
+		"AND rp.privilege_level IN ('d.t', 't') " +
+		"GROUP BY rp.role_name, tbl.reldatabase, tbl.relname, privilege_map.external_name"
 
 	InformationSchemaColumnPrivilegesDDL = "CREATE TABLE information_schema.`COLUMN_PRIVILEGES` (" +
 		"`GRANTEE` varchar(292) NOT NULL DEFAULT ''," +

pkg/util/sysview/predefined_test.go

@@ -172,6 +172,59 @@ func TestInformationSchemaCheckConstraintsDDL(t *testing.T) {
 	}
 }
 
+func TestInformationSchemaTablePrivilegesDDL(t *testing.T) {
+	assert.True(t, strings.HasPrefix(
+		InformationSchemaTablePrivilegesDDL,
+		"CREATE VIEW information_schema.`TABLE_PRIVILEGES` AS"))
+	for _, projection := range []string{
+		"CAST(rp.role_name AS varchar(292)) AS `GRANTEE`",
+		"CAST('def' AS varchar(512)) AS `TABLE_CATALOG`",
+		"CAST(tbl.reldatabase AS varchar(64)) AS `TABLE_SCHEMA`",
+		"CAST(tbl.relname AS varchar(64)) AS `TABLE_NAME`",
+		"CAST(privilege_map.external_name AS varchar(64)) AS `PRIVILEGE_TYPE`",
+		"max(case when rp.with_grant_option or rp.privilege_name = 'table ownership' then 1 else 0 end)",
+		"CAST(case when max(",
+	} {
+		assert.Contains(t, InformationSchemaTablePrivilegesDDL, projection)
+	}
+	for internalName, externalName := range map[string]string{
+		"select":    "SELECT",
+		"insert":    "INSERT",
+		"update":    "UPDATE",
+		"truncate":  "TRUNCATE",
+		"delete":    "DELETE",
+		"reference": "REFERENCES",
+		"index":     "INDEX",
+		"values":    "VALUES",
+	} {
+		assert.Contains(t, InformationSchemaTablePrivilegesDDL,
+			fmt.Sprintf("'%s'", internalName))
+		assert.Contains(t, InformationSchemaTablePrivilegesDDL,
+			fmt.Sprintf("'%s'", externalName))
+	}
+	assert.Contains(t, InformationSchemaTablePrivilegesDDL, "FROM mo_catalog.mo_role_privs rp")
+	assert.Contains(t, InformationSchemaTablePrivilegesDDL,
+		"JOIN mo_catalog.mo_tables tbl ON rp.obj_id = tbl.rel_id")
+	assert.Contains(t, InformationSchemaTablePrivilegesDDL,
+		"rp.privilege_name IN ('table all', 'table ownership')")
+	assert.Contains(t, InformationSchemaTablePrivilegesDDL,
+		"tbl.account_id = current_account_id()")
+	assert.Contains(t, InformationSchemaTablePrivilegesDDL,
+		"rp.obj_type IN ('table', 'view')")
+	assert.Contains(t, InformationSchemaTablePrivilegesDDL,
+		"rp.privilege_level IN ('d.t', 't')")
+	assert.Contains(t, InformationSchemaTablePrivilegesDDL,
+		"GROUP BY rp.role_name, tbl.reldatabase, tbl.relname, privilege_map.external_name")
+	assert.NotContains(t, InformationSchemaTablePrivilegesDDL,
+		"upper(rp.privilege_name)")
+
+	statements, err := mysql.Parse(context.Background(), InformationSchemaTablePrivilegesDDL, 1)
+	assert.NoError(t, err)
+	for _, statement := range statements {
+		statement.Free()
+	}
+}
+
 func TestInformationSchemaCharacterSetsData(t *testing.T) {
 	for _, expected := range []string{
 		"('binary','binary','Binary pseudo charset',1)",

test/distributed/cases/mo_cloud/mo_cloud.result

@@ -800,7 +800,7 @@ information_schema  ¦  schema_privileges  ¦  r  ¦  accountadmin  𝄀
 information_schema  ¦  schemata  ¦  v  ¦  accountadmin  𝄀
 information_schema  ¦  statistics  ¦  v  ¦  accountadmin  𝄀
 information_schema  ¦  table_constraints  ¦  v  ¦  accountadmin  𝄀
-information_schema  ¦  table_privileges  ¦  r  ¦  accountadmin  𝄀
+information_schema  ¦  table_privileges  ¦  v  ¦  accountadmin  𝄀
 information_schema  ¦  tables  ¦  v  ¦  accountadmin  𝄀
 information_schema  ¦  triggers  ¦  r  ¦  accountadmin  𝄀
 information_schema  ¦  user_privileges  ¦  r  ¦  accountadmin  𝄀
@@ -873,6 +873,7 @@ referential_constraints  ¦  accountadmin  𝄀
 schemata  ¦  accountadmin  𝄀
 statistics  ¦  accountadmin  𝄀
 table_constraints  ¦  accountadmin  𝄀
+table_privileges  ¦  accountadmin  𝄀
 tables  ¦  accountadmin  𝄀
 views  ¦  accountadmin
 SELECT relname AS `name`, IF (role_name IS NULL, '-', role_name) AS `owner` FROM mo_catalog.mo_tables LEFT JOIN mo_catalog.mo_role ON mo_catalog.mo_tables.owner=role_id WHERE relkind IN ('v') AND reldatabase='mo_sample_data_tpch_sf1';
@@ -906,7 +907,6 @@ parameters  ¦  0  ¦  0  ¦  accountadmin  𝄀
 profiling  ¦  0  ¦  0  ¦  accountadmin  𝄀
 routines  ¦  0  ¦  0  ¦  accountadmin  𝄀
 schema_privileges  ¦  0  ¦  0  ¦  accountadmin  𝄀
-table_privileges  ¦  0  ¦  0  ¦  accountadmin  𝄀
 triggers  ¦  0  ¦  0  ¦  accountadmin  𝄀
 user_privileges  ¦  0  ¦  0  ¦  accountadmin
 SELECT relname AS `name`, mo_table_rows(reldatabase, relname) AS `rows`, mo_table_size(reldatabase, relname) AS `size`, if (role_name IS NULL, '-', role_name) AS `owner` FROM mo_catalog.mo_tables LEFT JOIN mo_catalog.mo_role ON mo_catalog.mo_tables.owner=role_id WHERE relkind IN ('r','e','cluster') AND reldatabase='mo_sample_data_tpch_sf1';

test/distributed/cases/zz_accesscontrol/information_schema_table_privileges.result

@@ -0,0 +1,71 @@
+drop role if exists table_privileges_aggregate_role;
+drop role if exists table_privileges_reference_role;
+drop role if exists table_privileges_view_role;
+drop database if exists table_privileges_metadata_db;
+create database table_privileges_metadata_db;
+create table table_privileges_metadata_db.t(id int);
+create table table_privileges_metadata_db.reference_t(id int);
+create view table_privileges_metadata_db.v as select * from table_privileges_metadata_db.t;
+create role table_privileges_aggregate_role;
+create role table_privileges_reference_role;
+create role table_privileges_view_role;
+grant all on table table_privileges_metadata_db.t to table_privileges_aggregate_role;
+grant select on table table_privileges_metadata_db.t to table_privileges_aggregate_role with grant option;
+select privilege_type, is_grantable
+from information_schema.table_privileges
+where grantee = 'table_privileges_aggregate_role'
+and table_schema = 'table_privileges_metadata_db'
+and table_name = 't'
+order by privilege_type;
+➤ privilege_type[12,64,0]  ¦  is_grantable[12,3,0]  𝄀
+DELETE  ¦  NO  𝄀
+INDEX  ¦  NO  𝄀
+INSERT  ¦  NO  𝄀
+REFERENCES  ¦  NO  𝄀
+SELECT  ¦  YES  𝄀
+TRUNCATE  ¦  NO  𝄀
+UPDATE  ¦  NO  𝄀
+VALUES  ¦  NO
+grant ownership on table table_privileges_metadata_db.t to table_privileges_aggregate_role;
+select privilege_type, is_grantable
+from information_schema.table_privileges
+where grantee = 'table_privileges_aggregate_role'
+and table_schema = 'table_privileges_metadata_db'
+and table_name = 't'
+order by privilege_type;
+➤ privilege_type[12,64,0]  ¦  is_grantable[12,3,0]  𝄀
+DELETE  ¦  YES  𝄀
+INDEX  ¦  YES  𝄀
+INSERT  ¦  YES  𝄀
+REFERENCES  ¦  YES  𝄀
+SELECT  ¦  YES  𝄀
+TRUNCATE  ¦  YES  𝄀
+UPDATE  ¦  YES  𝄀
+VALUES  ¦  YES
+grant reference on table table_privileges_metadata_db.reference_t to table_privileges_reference_role;
+select grantee, table_catalog, table_schema, table_name, privilege_type, is_grantable
+from information_schema.table_privileges
+where grantee = 'table_privileges_reference_role'
+and table_schema = 'table_privileges_metadata_db'
+and table_name = 'reference_t';
+➤ grantee[12,292,0]  ¦  table_catalog[12,512,0]  ¦  table_schema[12,64,0]  ¦  table_name[12,64,0]  ¦  privilege_type[12,64,0]  ¦  is_grantable[12,3,0]  𝄀
+table_privileges_reference_role  ¦  def  ¦  table_privileges_metadata_db  ¦  reference_t  ¦  REFERENCES  ¦  NO
+grant select on view table_privileges_metadata_db.v to table_privileges_view_role;
+select grantee, table_catalog, table_schema, table_name, privilege_type, is_grantable
+from information_schema.table_privileges
+where grantee = 'table_privileges_view_role'
+and table_schema = 'table_privileges_metadata_db'
+and table_name = 'v';
+➤ grantee[12,292,0]  ¦  table_catalog[12,512,0]  ¦  table_schema[12,64,0]  ¦  table_name[12,64,0]  ¦  privilege_type[12,64,0]  ¦  is_grantable[12,3,0]  𝄀
+table_privileges_view_role  ¦  def  ¦  table_privileges_metadata_db  ¦  v  ¦  SELECT  ¦  NO
+revoke select on view table_privileges_metadata_db.v from table_privileges_view_role;
+select grantee, table_catalog, table_schema, table_name, privilege_type, is_grantable
+from information_schema.table_privileges
+where grantee = 'table_privileges_view_role'
+and table_schema = 'table_privileges_metadata_db'
+and table_name = 'v';
+➤ grantee[12,292,0]  ¦  table_catalog[12,512,0]  ¦  table_schema[12,64,0]  ¦  table_name[12,64,0]  ¦  privilege_type[12,64,0]  ¦  is_grantable[12,3,0]
+drop role table_privileges_aggregate_role;
+drop role table_privileges_reference_role;
+drop role table_privileges_view_role;
+drop database table_privileges_metadata_db;

test/distributed/cases/zz_accesscontrol/information_schema_table_privileges.sql

@@ -0,0 +1,54 @@
+drop role if exists table_privileges_aggregate_role;
+drop role if exists table_privileges_reference_role;
+drop role if exists table_privileges_view_role;
+drop database if exists table_privileges_metadata_db;
+
+create database table_privileges_metadata_db;
+create table table_privileges_metadata_db.t(id int);
+create table table_privileges_metadata_db.reference_t(id int);
+create view table_privileges_metadata_db.v as select * from table_privileges_metadata_db.t;
+create role table_privileges_aggregate_role;
+create role table_privileges_reference_role;
+create role table_privileges_view_role;
+
+grant all on table table_privileges_metadata_db.t to table_privileges_aggregate_role;
+grant select on table table_privileges_metadata_db.t to table_privileges_aggregate_role with grant option;
+select privilege_type, is_grantable
+from information_schema.table_privileges
+where grantee = 'table_privileges_aggregate_role'
+  and table_schema = 'table_privileges_metadata_db'
+  and table_name = 't'
+order by privilege_type;
+
+grant ownership on table table_privileges_metadata_db.t to table_privileges_aggregate_role;
+select privilege_type, is_grantable
+from information_schema.table_privileges
+where grantee = 'table_privileges_aggregate_role'
+  and table_schema = 'table_privileges_metadata_db'
+  and table_name = 't'
+order by privilege_type;
+
+grant reference on table table_privileges_metadata_db.reference_t to table_privileges_reference_role;
+select grantee, table_catalog, table_schema, table_name, privilege_type, is_grantable
+from information_schema.table_privileges
+where grantee = 'table_privileges_reference_role'
+  and table_schema = 'table_privileges_metadata_db'
+  and table_name = 'reference_t';
+
+grant select on view table_privileges_metadata_db.v to table_privileges_view_role;
+select grantee, table_catalog, table_schema, table_name, privilege_type, is_grantable
+from information_schema.table_privileges
+where grantee = 'table_privileges_view_role'
+  and table_schema = 'table_privileges_metadata_db'
+  and table_name = 'v';
+revoke select on view table_privileges_metadata_db.v from table_privileges_view_role;
+select grantee, table_catalog, table_schema, table_name, privilege_type, is_grantable
+from information_schema.table_privileges
+where grantee = 'table_privileges_view_role'
+  and table_schema = 'table_privileges_metadata_db'
+  and table_name = 'v';
+
+drop role table_privileges_aggregate_role;
+drop role table_privileges_reference_role;
+drop role table_privileges_view_role;
+drop database table_privileges_metadata_db;