fix!: forbid to change information_schema

src/operator/src/error.rs

@@ -269,6 +269,13 @@ pub enum Error {
         location: Location,
     },
 
+    #[snafu(display("Schema `{name}` is read-only"))]
+    SchemaReadOnly {
+        name: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
     #[snafu(display("Table occurs error"))]
     Table {
         #[snafu(implicit)]
@@ -718,6 +725,7 @@ impl ErrorExt for Error {
             | Error::SchemaNotFound { .. }
             | Error::SchemaExists { .. }
             | Error::SchemaInUse { .. }
+            | Error::SchemaReadOnly { .. }
             | Error::ColumnNotFound { .. }
             | Error::BuildRegex { .. }
             | Error::InvalidSchema { .. }

src/operator/src/req_convert/insert/stmt_to_region.rs

@@ -29,7 +29,7 @@ use table::TableRef;
 use crate::error::{
     CatalogSnafu, ColumnDataTypeSnafu, ColumnDefaultValueSnafu, ColumnNoneDefaultValueSnafu,
     ColumnNotFoundSnafu, InvalidSqlSnafu, MissingInsertBodySnafu, ParseSqlSnafu, Result,
-    TableNotFoundSnafu,
+    SchemaReadOnlySnafu, TableNotFoundSnafu,
 };
 use crate::req_convert::common::partitioner::Partitioner;
 use crate::req_convert::insert::semantic_type;
@@ -65,6 +65,11 @@ impl<'a> StatementToRegion<'a> {
         let table_schema = table.schema();
         let table_info = table.table_info();
 
+        ensure!(
+            schema != "information_schema",
+            SchemaReadOnlySnafu { name: schema }
+        );
+
         let column_names = column_names(stmt, &table_schema);
         let column_count = column_names.len();
 

src/operator/src/statement/ddl.rs

@@ -72,8 +72,8 @@ use crate::error::{
     ExtractTableNamesSnafu, FlowNotFoundSnafu, InvalidPartitionColumnsSnafu,
     InvalidPartitionRuleSnafu, InvalidPartitionSnafu, InvalidTableNameSnafu, InvalidViewNameSnafu,
     InvalidViewStmtSnafu, ParseSqlValueSnafu, Result, SchemaInUseSnafu, SchemaNotFoundSnafu,
-    SubstraitCodecSnafu, TableAlreadyExistsSnafu, TableMetadataManagerSnafu, TableNotFoundSnafu,
-    UnrecognizedTableOptionSnafu, ViewAlreadyExistsSnafu,
+    SchemaReadOnlySnafu, SubstraitCodecSnafu, TableAlreadyExistsSnafu, TableMetadataManagerSnafu,
+    TableNotFoundSnafu, UnrecognizedTableOptionSnafu, ViewAlreadyExistsSnafu,
 };
 use crate::expr_factory;
 use crate::statement::show::create_partitions_stmt;
@@ -151,6 +151,13 @@ impl StatementExecutor {
         partitions: Option<Partitions>,
         query_ctx: QueryContextRef,
     ) -> Result<TableRef> {
+        ensure!(
+            create_table.schema_name != "information_schema",
+            SchemaReadOnlySnafu {
+                name: create_table.schema_name.clone()
+            }
+        );
+
         // Check if is creating logical table
         if create_table.engine == METRIC_ENGINE_NAME
             && create_table
@@ -645,6 +652,13 @@ impl StatementExecutor {
     ) -> Result<Output> {
         let mut tables = Vec::with_capacity(table_names.len());
         for table_name in table_names {
+            ensure!(
+                table_name.schema_name != "information_schema",
+                SchemaReadOnlySnafu {
+                    name: table_name.schema_name.clone()
+                }
+            );
+
             if let Some(table) = self
                 .catalog_manager
                 .table(
@@ -694,6 +708,11 @@ impl StatementExecutor {
         drop_if_exists: bool,
         query_context: QueryContextRef,
     ) -> Result<Output> {
+        ensure!(
+            schema != "information_schema",
+            SchemaReadOnlySnafu { name: schema }
+        );
+
         if self
             .catalog_manager
             .schema_exists(&catalog, &schema)
@@ -725,6 +744,13 @@ impl StatementExecutor {
         table_name: TableName,
         query_context: QueryContextRef,
     ) -> Result<Output> {
+        ensure!(
+            table_name.schema_name != "information_schema",
+            SchemaReadOnlySnafu {
+                name: table_name.schema_name.clone()
+            }
+        );
+
         let table = self
             .catalog_manager
             .table(
@@ -794,6 +820,13 @@ impl StatementExecutor {
         expr: AlterExpr,
         query_context: QueryContextRef,
     ) -> Result<Output> {
+        ensure!(
+            expr.schema_name != "information_schema",
+            SchemaReadOnlySnafu {
+                name: expr.schema_name.clone()
+            }
+        );
+
         let catalog_name = if expr.catalog_name.is_empty() {
             DEFAULT_CATALOG_NAME.to_string()
         } else {

src/query/src/datafusion.rs

@@ -50,7 +50,7 @@ pub use crate::datafusion::planner::DfContextProviderAdapter;
 use crate::error::{
     CatalogSnafu, CreateRecordBatchSnafu, DataFusionSnafu, MissingTableMutationHandlerSnafu,
     MissingTimestampColumnSnafu, QueryExecutionSnafu, Result, TableMutationSnafu,
-    TableNotFoundSnafu, UnsupportedExprSnafu,
+    TableNotFoundSnafu, TableReadOnlySnafu, UnsupportedExprSnafu,
 };
 use crate::executor::QueryExecutor;
 use crate::logical_optimizer::LogicalOptimizer;
@@ -173,6 +173,12 @@ impl DatafusionQueryEngine {
         let schema_name = table_name.schema.to_string();
         let table_name = table_name.table.to_string();
         let table_schema = table.schema();
+
+        ensure!(
+            schema_name != "information_schema",
+            TableReadOnlySnafu { table: table_name }
+        );
+
         let ts_column = table_schema
             .timestamp_column()
             .map(|x| &x.name)
@@ -212,10 +218,19 @@ impl DatafusionQueryEngine {
         column_vectors: HashMap<String, VectorRef>,
         query_ctx: QueryContextRef,
     ) -> Result<Output> {
+        let catalog_name = table_name.catalog.to_string();
+        let schema_name = table_name.schema.to_string();
+        let table_name = table_name.table.to_string();
+
+        ensure!(
+            schema_name != "information_schema",
+            TableReadOnlySnafu { table: table_name }
+        );
+
         let request = InsertRequest {
-            catalog_name: table_name.catalog.to_string(),
-            schema_name: table_name.schema.to_string(),
-            table_name: table_name.table.to_string(),
+            catalog_name,
+            schema_name,
+            table_name,
             columns_values: column_vectors,
         };
 

src/query/src/error.rs

@@ -307,6 +307,13 @@ pub enum Error {
         location: Location,
         source: BoxedError,
     },
+
+    #[snafu(display("Cannot change read-only table: {}", table))]
+    TableReadOnly {
+        table: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
 }
 
 impl ErrorExt for Error {
@@ -320,6 +327,7 @@ impl ErrorExt for Error {
             UnsupportedExpr { .. }
             | Unimplemented { .. }
             | TableNotFound { .. }
+            | TableReadOnly { .. }
             | UnknownTable { .. }
             | TimeIndexNotFound { .. }
             | ParseTimestamp { .. }

tests/cases/standalone/common/information_schema/region_peers.result

@@ -1,5 +1,5 @@
 --- test information_schema.region_peers ----
-USE INFORMATION_SCHEMA;
+USE public;
 
 Affected Rows: 0
 
@@ -35,6 +35,10 @@ CREATE TABLE region_peers_test (
 
 Affected Rows: 0
 
+use INFORMATION_SCHEMA;
+
+Affected Rows: 0
+
 SELECT COUNT(distinct region_id) FROM region_peers;
 
 +----------------------------------------+
@@ -43,11 +47,11 @@ SELECT COUNT(distinct region_id) FROM region_peers;
 | 6                                      |
 +----------------------------------------+
 
-DROP TABLE region_peers_t1, region_peers_t2, region_peers_phy, region_peers_test;
+use public;
 
 Affected Rows: 0
 
-USE public;
+DROP TABLE region_peers_t1, region_peers_t2, region_peers_phy, region_peers_test;
 
 Affected Rows: 0
 

tests/cases/standalone/common/information_schema/region_peers.sql

@@ -1,5 +1,5 @@
 --- test information_schema.region_peers ----
-USE INFORMATION_SCHEMA;
+USE public;
 
 CREATE TABLE region_peers_phy (ts timestamp time index, val double) engine = metric with ("physical_metric_table" = "");
 
@@ -25,10 +25,10 @@ CREATE TABLE region_peers_test (
     a >= 20,
 );
 
-SELECT COUNT(distinct region_id) FROM region_peers;
-
-DROP TABLE region_peers_t1, region_peers_t2, region_peers_phy, region_peers_test;
+use INFORMATION_SCHEMA;
 
-USE public;
+SELECT COUNT(distinct region_id) FROM region_peers;
 
+use public;
 
+DROP TABLE region_peers_t1, region_peers_t2, region_peers_phy, region_peers_test;

tests/cases/standalone/common/information_schema/table_constraints.result

@@ -25,34 +25,42 @@ SELECT * FROM TABLE_CONSTRAINTS ORDER BY TABLE_NAME, CONSTRAINT_NAME;
 | def                | public            | PRIMARY         | public       | numbers    | PRIMARY KEY     | YES      |
 +--------------------+-------------------+-----------------+--------------+------------+-----------------+----------+
 
+use public;
+
+Affected Rows: 0
+
 CREATE TABLE test(i double, j string, ts timestamp time index, primary key(j));
 
 Affected Rows: 0
 
+use INFORMATION_SCHEMA;
+
+Affected Rows: 0
+
 SELECT * FROM TABLE_CONSTRAINTS ORDER BY TABLE_NAME, CONSTRAINT_NAME;
 
-+--------------------+--------------------+-----------------+--------------------+------------+-----------------+----------+
-| constraint_catalog | constraint_schema  | constraint_name | table_schema       | table_name | constraint_type | enforced |
-+--------------------+--------------------+-----------------+--------------------+------------+-----------------+----------+
-| def                | public             | PRIMARY         | public             | numbers    | PRIMARY KEY     | YES      |
-| def                | information_schema | PRIMARY         | information_schema | test       | PRIMARY KEY     | YES      |
-| def                | information_schema | TIME INDEX      | information_schema | test       | TIME INDEX      | YES      |
-+--------------------+--------------------+-----------------+--------------------+------------+-----------------+----------+
++--------------------+-------------------+-----------------+--------------+------------+-----------------+----------+
+| constraint_catalog | constraint_schema | constraint_name | table_schema | table_name | constraint_type | enforced |
++--------------------+-------------------+-----------------+--------------+------------+-----------------+----------+
+| def                | public            | PRIMARY         | public       | numbers    | PRIMARY KEY     | YES      |
+| def                | public            | PRIMARY         | public       | test       | PRIMARY KEY     | YES      |
+| def                | public            | TIME INDEX      | public       | test       | TIME INDEX      | YES      |
++--------------------+-------------------+-----------------+--------------+------------+-----------------+----------+
 
 SELECT * FROM TABLE_CONSTRAINTS WHERE TABLE_NAME = 'test' ORDER BY TABLE_NAME, CONSTRAINT_NAME;
 
-+--------------------+--------------------+-----------------+--------------------+------------+-----------------+----------+
-| constraint_catalog | constraint_schema  | constraint_name | table_schema       | table_name | constraint_type | enforced |
-+--------------------+--------------------+-----------------+--------------------+------------+-----------------+----------+
-| def                | information_schema | PRIMARY         | information_schema | test       | PRIMARY KEY     | YES      |
-| def                | information_schema | TIME INDEX      | information_schema | test       | TIME INDEX      | YES      |
-+--------------------+--------------------+-----------------+--------------------+------------+-----------------+----------+
++--------------------+-------------------+-----------------+--------------+------------+-----------------+----------+
+| constraint_catalog | constraint_schema | constraint_name | table_schema | table_name | constraint_type | enforced |
++--------------------+-------------------+-----------------+--------------+------------+-----------------+----------+
+| def                | public            | PRIMARY         | public       | test       | PRIMARY KEY     | YES      |
+| def                | public            | TIME INDEX      | public       | test       | TIME INDEX      | YES      |
++--------------------+-------------------+-----------------+--------------+------------+-----------------+----------+
 
-DROP TABLE test;
+use public;
 
 Affected Rows: 0
 
-USE public;
+DROP TABLE test;
 
 Affected Rows: 0
 

tests/cases/standalone/common/information_schema/table_constraints.sql

@@ -6,12 +6,16 @@ DESC TABLE TABLE_CONSTRAINTS;
 
 SELECT * FROM TABLE_CONSTRAINTS ORDER BY TABLE_NAME, CONSTRAINT_NAME;
 
+use public;
+
 CREATE TABLE test(i double, j string, ts timestamp time index, primary key(j));
 
+use INFORMATION_SCHEMA;
+
 SELECT * FROM TABLE_CONSTRAINTS ORDER BY TABLE_NAME, CONSTRAINT_NAME;
 
 SELECT * FROM TABLE_CONSTRAINTS WHERE TABLE_NAME = 'test' ORDER BY TABLE_NAME, CONSTRAINT_NAME;
 
-DROP TABLE test;
+use public;
 
-USE public;
+DROP TABLE test;

tests/cases/standalone/common/information_schema/tables.result

@@ -44,3 +44,7 @@ drop schema abcde;
 
 Affected Rows: 0
 
+drop schema information_schema;
+
+Error: 1004(InvalidArguments), Schema `information_schema` is read-only
+

tests/cases/standalone/common/information_schema/tables.sql

@@ -17,3 +17,5 @@ use public;
 drop schema abc;
 
 drop schema abcde;
+
+drop schema information_schema;