fix: correct information schema views metadata

pkg/bootstrap/versions/v4_0_6/tenant_upgrade_list.go

@@ -49,6 +49,7 @@ var tenantUpgEntries = []versions.UpgradeEntry{
 	upgradeInformationSchemaCollationCharacterSetApplicability(),
 	backfillMoColumnsAttIsUnsigned(),
 	upgradeInformationSchemaStatistics(),
+	upgradeInformationSchemaViews(),
 }
 
 const moColumnsUnsignedMismatchPredicate = "account_id = current_account_id() " +
@@ -96,6 +97,19 @@ func upgradeInformationSchemaColumnsHideInternalColumns() versions.UpgradeEntry
 	return upgradeInformationSchemaColumns()
 }
 
+// Keep a separate entry so existing tenants replace the historical VIEWS
+// definition that exposed CREATE VIEW text and claimed every view was writable.
+func upgradeInformationSchemaViews() versions.UpgradeEntry {
+	return versions.UpgradeEntry{
+		Schema:    sysview.InformationDBConst,
+		TableName: "VIEWS",
+		UpgType:   versions.MODIFY_VIEW,
+		UpgSql:    sysview.InformationSchemaViewsDDL,
+		CheckFunc: checkViewDefinition("VIEWS", sysview.InformationSchemaViewsDDL),
+		PreSql:    fmt.Sprintf("DROP VIEW IF EXISTS %s.VIEWS;", sysview.InformationDBConst),
+	}
+}
+
 func addForeignKeyMetadataColumn(column, definition, after string) versions.UpgradeEntry {
 	return versions.UpgradeEntry{
 		Schema:    catalog.MO_CATALOG,

pkg/bootstrap/versions/v4_0_6/upgrade_test.go

@@ -37,7 +37,7 @@ import (
 )
 
 func TestUpgradeEntries(t *testing.T) {
-	require.Len(t, tenantUpgEntries, 21)
+	require.Len(t, tenantUpgEntries, 22)
 	require.Len(t, clusterUpgEntries, 3)
 	require.Equal(t, retireKafkaSinkDaemonTasks.UpgSql, clusterUpgEntries[0].UpgSql)
 	require.Equal(t, catalog.MO_VIEW_DEPENDENCIES, clusterUpgEntries[1].TableName)
@@ -139,6 +139,12 @@ func TestUpgradeEntries(t *testing.T) {
 	require.Equal(t, sysview.InformationSchemaStatisticsDDL, statistics.UpgSql)
 	require.Contains(t, strings.ToLower(statistics.PreSql),
 		"drop view if exists information_schema.statistics")
+	views := tenantUpgEntries[21]
+	require.Equal(t, versions.MODIFY_VIEW, views.UpgType)
+	require.Equal(t, sysview.InformationDBConst, views.Schema)
+	require.Equal(t, "VIEWS", views.TableName)
+	require.Equal(t, sysview.InformationSchemaViewsDDL, views.UpgSql)
+	require.Contains(t, strings.ToLower(views.PreSql), "drop view if exists information_schema.views")
 }
 
 func TestMoColumnsUnsignedBackfillPredicate(t *testing.T) {
@@ -247,7 +253,7 @@ func TestUserDefinedFunctionArgumentTypesBackfillRejectsOversizedSignature(t *te
 }
 
 func TestForeignKeyMetadataTenantUpgradeEntries(t *testing.T) {
-	require.Len(t, tenantUpgEntries, 21)
+	require.Len(t, tenantUpgEntries, 22)
 
 	for i, column := range []string{"referenced_index_name", "on_delete_origin", "on_update_origin"} {
 		entry := tenantUpgEntries[2+i]
@@ -486,6 +492,7 @@ func TestTenantViewDefinitionChecks(t *testing.T) {
 		upgradeInformationSchemaTableConstraints(),
 		upgradeInformationSchemaCollationCharacterSetApplicability(),
 		upgradeInformationSchemaStatistics(),
+		upgradeInformationSchemaViews(),
 	}
 
 	for _, entry := range entries {
@@ -625,6 +632,8 @@ func TestVersionHandleLifecycleWithNoLegacyDefinitions(t *testing.T) {
 				return true, sysview.InformationSchemaColumnsDDL, nil
 			case "STATISTICS":
 				return true, sysview.InformationSchemaStatisticsDDL, nil
+			case "VIEWS":
+				return true, sysview.InformationSchemaViewsDDL, nil
 			default:
 				return false, "", errors.New("unexpected view")
 			}

pkg/util/sysview/predefined.go

@@ -21,6 +21,45 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/catalog"
 )
 
+const (
+	informationSchemaViewIdentifierPattern = "(?:`(?:``|[^`])*`|\"(?:\"\"|[^\"])*\"|[^[:space:].(),]+)"
+	// GetRootSql preserves line comments, so separators in the persisted DDL must
+	// accept every lexer-supported form wherever valid SQL permits whitespace
+	// between view tokens.
+	informationSchemaViewLineCommentPattern       = "(?:(?:--|#|//)[^\\r\\n]*(?:\\r?\\n|$))"
+	informationSchemaViewOptionalSeparatorPattern = "(?:[[:space:]]|" + informationSchemaViewLineCommentPattern + ")*"
+	informationSchemaViewRequiredSeparatorPattern = "(?:[[:space:]]|" + informationSchemaViewLineCommentPattern + ")+"
+	// The non-greedy span before VIEW covers MatrixOne's supported ALGORITHM,
+	// DEFINER, and SQL SECURITY clauses as well as mysqldump's version comments.
+	informationSchemaViewDefinitionPrefixPattern = "(?is)^[[:space:]]*(?:/[*]![0-9]+[[:space:]]*)?" +
+		"(?:create(?:" + informationSchemaViewRequiredSeparatorPattern + "or" + informationSchemaViewRequiredSeparatorPattern + "replace)?|alter).*?" + informationSchemaViewRequiredSeparatorPattern + "view" + informationSchemaViewRequiredSeparatorPattern +
+		"(?:if" + informationSchemaViewRequiredSeparatorPattern + "(?:not" + informationSchemaViewRequiredSeparatorPattern + ")?exists" + informationSchemaViewRequiredSeparatorPattern + ")?" +
+		informationSchemaViewIdentifierPattern +
+		"(?:" + informationSchemaViewOptionalSeparatorPattern + "[.]" + informationSchemaViewOptionalSeparatorPattern + informationSchemaViewIdentifierPattern + ")?" +
+		informationSchemaViewOptionalSeparatorPattern + "(?:[(]" + informationSchemaViewOptionalSeparatorPattern + informationSchemaViewIdentifierPattern +
+		"(?:" + informationSchemaViewOptionalSeparatorPattern + "[,]" + informationSchemaViewOptionalSeparatorPattern + informationSchemaViewIdentifierPattern + ")*" + informationSchemaViewOptionalSeparatorPattern + "[)])?" +
+		informationSchemaViewRequiredSeparatorPattern + "as" + informationSchemaViewRequiredSeparatorPattern
+	informationSchemaViewDefinitionVersionCommentPrefixPattern = "(?is)^[[:space:]]*/[*]![0-9]+[[:space:]]*"
+	informationSchemaViewDefinitionCommentSuffixPattern        = "(?is)[[:space:]]*[*]/[[:space:]]*;?[[:space:]]*$"
+	informationSchemaViewStatementSQL                          = "coalesce(json_extract_string(tbl.viewdef, '$.Stmt'), tbl.rel_createsql)"
+	informationSchemaViewStatementWithoutTerminatorSQL         = "trim(regexp_replace(trim(" +
+		informationSchemaViewStatementSQL + "), '[;][[:space:]]*$', '', 1, 1))"
+	informationSchemaViewDefinitionPrefixLengthSQL = "char_length(coalesce(regexp_substr(" +
+		informationSchemaViewStatementWithoutTerminatorSQL + ", '" + informationSchemaViewDefinitionPrefixPattern + "'), ''))"
+	informationSchemaViewDefinitionVersionCommentPrefixLengthSQL = "char_length(coalesce(regexp_substr(" +
+		informationSchemaViewStatementWithoutTerminatorSQL + ", '" + informationSchemaViewDefinitionVersionCommentPrefixPattern + "'), ''))"
+	// Keep the persisted system-view definition free of CASE/IF, which the
+	// database-clone catalog restore cannot parse in this view definition.
+	// Prefix lengths are counted in characters so they match substr even for
+	// multibyte view identifiers. The version-comment prefix recognizes only a
+	// mysqldump wrapper, so a trailing */ is removed only for that wrapper and
+	// not for an application comment.
+	informationSchemaViewDefinitionSQL = "cast(trim(substr(" + informationSchemaViewStatementWithoutTerminatorSQL +
+		", " + informationSchemaViewDefinitionPrefixLengthSQL + " + 1, char_length(" +
+		informationSchemaViewStatementWithoutTerminatorSQL + ") - " + informationSchemaViewDefinitionPrefixLengthSQL + " - " +
+		"2 * least(" + informationSchemaViewDefinitionVersionCommentPrefixLengthSQL + ", 1))) as text)"
+)
+
 // `mysql` database system tables
 // They are all Tenant level system tables
 var (
@@ -398,9 +437,9 @@ var (
 		"SELECT 'def' AS `TABLE_CATALOG`," +
 		"tbl.reldatabase AS `TABLE_SCHEMA`," +
 		"tbl.relname AS `TABLE_NAME`," +
-		"tbl.rel_createsql AS `VIEW_DEFINITION`," +
+		informationSchemaViewDefinitionSQL + " AS `VIEW_DEFINITION`," +
 		"'NONE' AS `CHECK_OPTION`," +
-		"'YES' AS `IS_UPDATABLE`," +
+		"'NO' AS `IS_UPDATABLE`," +
 		"usr.user_name + '@' + usr.user_host AS `DEFINER`," +
 		"'DEFINER' AS `SECURITY_TYPE`," +
 		"'utf8mb4' AS `CHARACTER_SET_CLIENT`," +

pkg/util/sysview/predefined_test.go

@@ -17,6 +17,7 @@ package sysview
 import (
 	"context"
 	"fmt"
+	"regexp"
 	"strings"
 	"testing"
 
@@ -229,6 +230,107 @@ func TestInformationSchemaCharacterSetsData(t *testing.T) {
 	assert.Equal(t, ddlIndex+1, dataIndex)
 }
 
+func TestInformationSchemaViewsMetadata(t *testing.T) {
+	assert.Contains(t, InformationSchemaViewsDDL,
+		"char_length(coalesce(regexp_substr(trim(regexp_replace(trim(coalesce(json_extract_string(tbl.viewdef, '$.Stmt'), tbl.rel_createsql))")
+	assert.Contains(t, InformationSchemaViewsDDL, "2 * least(char_length(coalesce(regexp_substr(")
+	assert.NotContains(t, InformationSchemaViewsDDL, "case when")
+	assert.NotContains(t, InformationSchemaViewsDDL, "trim(if(")
+	assert.Contains(t, InformationSchemaViewsDDL, "'NO' AS `IS_UPDATABLE`")
+	assert.NotContains(t, InformationSchemaViewsDDL, "tbl.rel_createsql AS `VIEW_DEFINITION`")
+
+	prefix := regexp.MustCompile(informationSchemaViewDefinitionPrefixPattern)
+	tests := []struct {
+		name       string
+		createSQL  string
+		definition string
+	}{
+		{
+			name:       "aggregate view",
+			createSQL:  "create view agg_v as select a, count(*) cnt from t group by a;",
+			definition: "select a, count(*) cnt from t group by a",
+		},
+		{
+			name:       "qualified stable view",
+			createSQL:  "create view `db`.`v` as select `t`.`a` as `a` from `db`.`t`",
+			definition: "select `t`.`a` as `a` from `db`.`t`",
+		},
+		{
+			name:       "replace view with cte",
+			createSQL:  "CREATE OR REPLACE VIEW IF NOT EXISTS \"db\".\"v as quoted\" AS WITH c AS (SELECT 1) SELECT * FROM c",
+			definition: "WITH c AS (SELECT 1) SELECT * FROM c",
+		},
+		{
+			name:       "alter view with explicit columns",
+			createSQL:  "  ALTER VIEW IF EXISTS `v` (`c as quoted`, plain) AS SELECT a AS plain, b FROM t",
+			definition: "SELECT a AS plain, b FROM t",
+		},
+		{
+			name:       "view options",
+			createSQL:  "CREATE ALGORITHM=MERGE DEFINER=`root`@`%` SQL SECURITY DEFINER VIEW `v` AS SELECT 1;",
+			definition: "SELECT 1",
+		},
+		{
+			name: "mysqldump version comments",
+			createSQL: "/*!50001 CREATE ALGORITHM=UNDEFINED *//*!50013 DEFINER=`root`@`%` SQL SECURITY DEFINER */" +
+				"/*!50001 VIEW `v` AS select 1 */;",
+			definition: "select 1",
+		},
+		{
+			name:       "select block comment remains intact",
+			createSQL:  "create view v as select 1 /* application comment */;",
+			definition: "select 1 /* application comment */",
+		},
+		{
+			name:       "line comment before as",
+			createSQL:  "create view v -- migration comment\n as select 1;",
+			definition: "select 1",
+		},
+		{
+			name:       "hash line comment before as",
+			createSQL:  "create view v # migration comment\n as select 1;",
+			definition: "select 1",
+		},
+		{
+			name:       "slash line comment before as",
+			createSQL:  "create view v // migration comment\n as select 1;",
+			definition: "select 1",
+		},
+		{
+			name:       "unrecognized metadata remains visible",
+			createSQL:  "select 1",
+			definition: "select 1",
+		},
+	}
+	suffix := regexp.MustCompile(informationSchemaViewDefinitionCommentSuffixPattern)
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			definition := strings.TrimSpace(prefix.ReplaceAllString(test.createSQL, ""))
+			if strings.HasPrefix(strings.TrimSpace(test.createSQL), "/*!") {
+				definition = strings.TrimSpace(suffix.ReplaceAllString(definition, ""))
+			}
+			definition = strings.TrimSuffix(definition, ";")
+			assert.Equal(t, test.definition, definition)
+		})
+	}
+	for _, createSQL := range []string{
+		"create view hash_comment_v # migration comment\n as select 1;",
+		"create view slash_comment_v // migration comment\n as select 1;",
+	} {
+		statements, err := mysql.Parse(context.Background(), createSQL, 1)
+		assert.NoError(t, err)
+		for _, statement := range statements {
+			statement.Free()
+		}
+	}
+
+	statements, err := mysql.Parse(context.Background(), InformationSchemaViewsDDL, 1)
+	assert.NoError(t, err)
+	for _, statement := range statements {
+		statement.Free()
+	}
+}
+
 func TestInformationSchemaDefaultCollationsMatchCanonicalDefinitions(t *testing.T) {
 	assert.Empty(t, DefaultCollationForCharset("unknown_charset"))
 	for _, charset := range []string{"binary", "utf8", "utf8mb4"} {

test/distributed/cases/system_variable/system_variables.result

@@ -259,7 +259,7 @@ table_schema  ¦  VARCHAR(5000)  ¦  YES  ¦    ¦  null  ¦    ¦    𝄀
 table_name  ¦  VARCHAR(5000)  ¦  YES  ¦    ¦  null  ¦    ¦    𝄀
 view_definition  ¦  TEXT(0)  ¦  YES  ¦    ¦  null  ¦    ¦    𝄀
 check_option  ¦  VARCHAR(4)  ¦  NO  ¦    ¦  null  ¦    ¦    𝄀
-is_updatable  ¦  VARCHAR(3)  ¦  NO  ¦    ¦  null  ¦    ¦    𝄀
+is_updatable  ¦  VARCHAR(2)  ¦  NO  ¦    ¦  null  ¦    ¦    𝄀
 definer  ¦  VARCHAR(65535)  ¦  YES  ¦    ¦  null  ¦    ¦    𝄀
 security_type  ¦  VARCHAR(7)  ¦  NO  ¦    ¦  null  ¦    ¦    𝄀
 character_set_client  ¦  VARCHAR(7)  ¦  NO  ¦    ¦  null  ¦    ¦    𝄀

test/distributed/cases/view/information_schema_views_metadata.result

@@ -0,0 +1,33 @@
+drop database if exists information_schema_views_metadata;
+create database information_schema_views_metadata;
+use information_schema_views_metadata;
+create table t(a int, b int);
+insert into t values (1, 10), (1, 20), (2, 30);
+create view direct_v as select a, b from t;
+create view agg_v as select a, count(*) cnt from t group by a;
+/*!50001 CREATE DEFINER = `root`@`%` VIEW dump_v AS select a from t */;
+create view line_comment_v -- migration-generated view
+as select a from t;
+create view hash_comment_v # migration-generated view
+as select a from t;
+create view slash_comment_v // migration-generated view
+as select a from t;
+select table_name, view_definition, is_updatable
+from information_schema.views
+where table_schema = 'information_schema_views_metadata'
+order by table_name;
+➤ table_name[12,5000,0]  ¦  view_definition[-1,16383,0]  ¦  is_updatable[12,2,0]  𝄀
+agg_v  ¦  select a, count(*) cnt from t group by a  ¦  NO  𝄀
+direct_v  ¦  select a, b from t  ¦  NO  𝄀
+dump_v  ¦  select a from t  ¦  NO  𝄀
+hash_comment_v  ¦  select a from t  ¦  NO  𝄀
+line_comment_v  ¦  select a from t  ¦  NO  𝄀
+slash_comment_v  ¦  select a from t  ¦  NO
+update agg_v set cnt = 1;
+invalid input: cannot insert/update/delete from view
+update direct_v set b = 1;
+invalid input: cannot insert/update/delete from view
+drop database information_schema_views_metadata;
+drop database if exists information_schema_views_clone;
+create database information_schema_views_clone clone information_schema;
+drop database information_schema_views_clone;

test/distributed/cases/view/information_schema_views_metadata.sql

@@ -0,0 +1,31 @@
+-- @label:bvt
+drop database if exists information_schema_views_metadata;
+create database information_schema_views_metadata;
+use information_schema_views_metadata;
+
+create table t(a int, b int);
+insert into t values (1, 10), (1, 20), (2, 30);
+create view direct_v as select a, b from t;
+create view agg_v as select a, count(*) cnt from t group by a;
+/*!50001 CREATE DEFINER = `root`@`%` VIEW dump_v AS select a from t */;
+create view line_comment_v -- migration-generated view
+as select a from t;
+create view hash_comment_v # migration-generated view
+as select a from t;
+create view slash_comment_v // migration-generated view
+as select a from t;
+
+select table_name, view_definition, is_updatable
+from information_schema.views
+where table_schema = 'information_schema_views_metadata'
+order by table_name;
+
+update agg_v set cnt = 1;
+update direct_v set b = 1;
+
+drop database information_schema_views_metadata;
+
+-- The stored VIEWS definition must remain executable when a system database is cloned.
+drop database if exists information_schema_views_clone;
+create database information_schema_views_clone clone information_schema;
+drop database information_schema_views_clone;

test/distributed/cases/zz_accesscontrol/inner_object.result

@@ -333,8 +333,8 @@ ac_db  ¦  ac_t1
 select count(*),table_name from information_schema.tables group by table_name having count(*) >1;
 ➤ count(*)[-5,64,0]  ¦  table_name[12,-1,0]
 select * from information_schema.views where table_name='ac_v1';
-➤ TABLE_CATALOG[12,2,0]  ¦  TABLE_SCHEMA[12,3750,0]  ¦  TABLE_NAME[12,3750,0]  ¦  VIEW_DEFINITION[12,0,0]  ¦  CHECK_OPTION[12,3,0]  ¦  IS_UPDATABLE[12,2,0]  ¦  DEFINER[12,49151,0]  ¦  SECURITY_TYPE[12,5,0]  ¦  CHARACTER_SET_CLIENT[12,5,0]  ¦  COLLATION_CONNECTION[12,13,0]  𝄀
-def  ¦  ac_db  ¦  ac_v1  ¦  create view `ac_db`.`ac_v1` as select `ac_t1`.`c1` as `c1` from `ac_db`.`ac_t1`  ¦  NONE  ¦  YES  ¦  admin@localhost  ¦  DEFINER  ¦  utf8mb4  ¦  utf8mb4_general_ci
+➤ TABLE_CATALOG[12,2,0]  ¦  TABLE_SCHEMA[12,3750,0]  ¦  TABLE_NAME[12,3750,0]  ¦  VIEW_DEFINITION[-1,16383,0]  ¦  CHECK_OPTION[12,3,0]  ¦  IS_UPDATABLE[12,2,0]  ¦  DEFINER[12,49151,0]  ¦  SECURITY_TYPE[12,5,0]  ¦  CHARACTER_SET_CLIENT[12,5,0]  ¦  COLLATION_CONNECTION[12,13,0]  𝄀
+def  ¦  ac_db  ¦  ac_v1  ¦  select `ac_t1`.`c1` as `c1` from `ac_db`.`ac_t1`  ¦  NONE  ¦  NO  ¦  admin@localhost  ¦  DEFINER  ¦  utf8mb4  ¦  utf8mb4_general_ci
 select * from information_schema.views where table_name='sys_v1';
 ➤ TABLE_CATALOG[12,2,0]  ¦  TABLE_SCHEMA[12,3750,0]  ¦  TABLE_NAME[12,3750,0]  ¦  VIEW_DEFINITION[12,0,0]  ¦  CHECK_OPTION[12,3,0]  ¦  IS_UPDATABLE[12,2,0]  ¦  DEFINER[12,49151,0]  ¦  SECURITY_TYPE[12,5,0]  ¦  CHARACTER_SET_CLIENT[12,5,0]  ¦  COLLATION_CONNECTION[12,13,0]
 select count(*),table_name from information_schema.views group by table_name having count(*)>1;
@@ -367,8 +367,8 @@ select table_schema,table_name from information_schema.tables where table_name='
 select count(*),table_name from information_schema.tables group by table_name having count(*) >1;
 ➤ count(*)[-5,64,0]  ¦  table_name[12,-1,0]
 select * from information_schema.views where table_name='sys_v1';
-➤ TABLE_CATALOG[12,2,0]  ¦  TABLE_SCHEMA[12,3750,0]  ¦  TABLE_NAME[12,3750,0]  ¦  VIEW_DEFINITION[12,0,0]  ¦  CHECK_OPTION[12,3,0]  ¦  IS_UPDATABLE[12,2,0]  ¦  DEFINER[12,49151,0]  ¦  SECURITY_TYPE[12,5,0]  ¦  CHARACTER_SET_CLIENT[12,5,0]  ¦  COLLATION_CONNECTION[12,13,0]  𝄀
-def  ¦  sys_db1  ¦  sys_v1  ¦  create view `sys_db1`.`sys_v1` as select `sys_t1`.`c1` as `c1` from `sys_db1`.`sys_t1`  ¦  NONE  ¦  YES  ¦  dump@localhost  ¦  DEFINER  ¦  utf8mb4  ¦  utf8mb4_general_ci
+➤ TABLE_CATALOG[12,2,0]  ¦  TABLE_SCHEMA[12,3750,0]  ¦  TABLE_NAME[12,3750,0]  ¦  VIEW_DEFINITION[-1,16383,0]  ¦  CHECK_OPTION[12,3,0]  ¦  IS_UPDATABLE[12,2,0]  ¦  DEFINER[12,49151,0]  ¦  SECURITY_TYPE[12,5,0]  ¦  CHARACTER_SET_CLIENT[12,5,0]  ¦  COLLATION_CONNECTION[12,13,0]  𝄀
+def  ¦  sys_db1  ¦  sys_v1  ¦  select `sys_t1`.`c1` as `c1` from `sys_db1`.`sys_t1`  ¦  NONE  ¦  NO  ¦  dump@localhost  ¦  DEFINER  ¦  utf8mb4  ¦  utf8mb4_general_ci
 select * from information_schema.views where table_name='ac_v1';
 ➤ TABLE_CATALOG[12,2,0]  ¦  TABLE_SCHEMA[12,3750,0]  ¦  TABLE_NAME[12,3750,0]  ¦  VIEW_DEFINITION[12,0,0]  ¦  CHECK_OPTION[12,3,0]  ¦  IS_UPDATABLE[12,2,0]  ¦  DEFINER[12,49151,0]  ¦  SECURITY_TYPE[12,5,0]  ¦  CHARACTER_SET_CLIENT[12,5,0]  ¦  COLLATION_CONNECTION[12,13,0]
 select count(*),table_name from information_schema.views group by table_name having count(*)>1;