fix: validate create table request in mito engine (#690)

* fix: validate create table request in mito engine

* fix: comment

* chore: remove TIMESTAMP_INDEX in system.rs

src/catalog/src/system.rs

@@ -43,7 +43,6 @@ use crate::error::{
 
 pub const ENTRY_TYPE_INDEX: usize = 0;
 pub const KEY_INDEX: usize = 1;
-pub const TIMESTAMP_INDEX: usize = 2;
 pub const VALUE_INDEX: usize = 3;
 
 pub struct SystemCatalogTable {
@@ -111,7 +110,7 @@ impl SystemCatalogTable {
                 desc: Some("System catalog table".to_string()),
                 schema: schema.clone(),
                 region_numbers: vec![0],
-                primary_key_indices: vec![ENTRY_TYPE_INDEX, KEY_INDEX, TIMESTAMP_INDEX],
+                primary_key_indices: vec![ENTRY_TYPE_INDEX, KEY_INDEX],
                 create_if_not_exists: true,
                 table_options: HashMap::new(),
             };

src/datanode/src/sql/create.rs

@@ -183,14 +183,6 @@ impl SqlHandler {
 
         ensure!(ts_index != usize::MAX, error::MissingTimestampColumnSnafu);
 
-        if primary_keys.is_empty() {
-            info!(
-                "Creating table: {} with time index column: {} upon primary keys absent",
-                table_ref, ts_index
-            );
-            primary_keys.push(ts_index);
-        }
-
         let columns_schemas: Vec<_> = stmt
             .columns
             .iter()
@@ -288,7 +280,6 @@ mod tests {
         assert_matches!(error, Error::MissingTimestampColumn { .. });
     }
 
-    /// If primary key is not specified, time index should be used as primary key.
     #[tokio::test]
     pub async fn test_primary_key_not_specified() {
         let handler = create_mock_sql_handler().await;
@@ -304,11 +295,8 @@ mod tests {
         let c = handler
             .create_to_request(42, parsed_stmt, TableReference::bare("demo_table"))
             .unwrap();
-        assert_eq!(1, c.primary_key_indices.len());
-        assert_eq!(
-            c.schema.timestamp_index().unwrap(),
-            c.primary_key_indices[0]
-        );
+        assert!(c.primary_key_indices.is_empty());
+        assert_eq!(c.schema.timestamp_index(), Some(1));
     }
 
     /// Constraints specified, not column cannot be found.

src/datanode/src/tests/test_util.rs

@@ -89,7 +89,7 @@ pub async fn create_test_table(
                         .expect("ts is expected to be timestamp column"),
                 ),
                 create_if_not_exists: true,
-                primary_key_indices: vec![3, 0], // "host" and "ts" are primary keys
+                primary_key_indices: vec![0], // "host" is in primary keys
                 table_options: HashMap::new(),
                 region_numbers: vec![0],
             },

src/frontend/src/instance.rs

@@ -825,7 +825,7 @@ mod tests {
                             memory DOUBLE NULL,
                             disk_util DOUBLE DEFAULT 9.9,
                             TIME INDEX (ts),
-                            PRIMARY KEY(ts, host)
+                            PRIMARY KEY(host)
                         ) engine=mito with(regions=1);"#;
         let output = SqlQueryHandler::do_query(&*instance, sql, query_ctx.clone())
             .await
@@ -1077,7 +1077,7 @@ mod tests {
             desc: None,
             column_defs,
             time_index: "ts".to_string(),
-            primary_keys: vec!["ts".to_string(), "host".to_string()],
+            primary_keys: vec!["host".to_string()],
             create_if_not_exists: true,
             table_options: Default::default(),
             table_id: None,

src/mito/src/engine.rs

@@ -21,7 +21,7 @@ use common_error::ext::BoxedError;
 use common_telemetry::logging;
 use datatypes::schema::SchemaRef;
 use object_store::ObjectStore;
-use snafu::{OptionExt, ResultExt};
+use snafu::{ensure, OptionExt, ResultExt};
 use store_api::storage::{
     ColumnDescriptorBuilder, ColumnFamilyDescriptor, ColumnFamilyDescriptorBuilder, ColumnId,
     CreateOptions, EngineContext as StorageEngineContext, OpenOptions, RegionDescriptorBuilder,
@@ -37,7 +37,8 @@ use tokio::sync::Mutex;
 use crate::config::EngineConfig;
 use crate::error::{
     self, BuildColumnDescriptorSnafu, BuildColumnFamilyDescriptorSnafu, BuildRegionDescriptorSnafu,
-    BuildRowKeyDescriptorSnafu, MissingTimestampIndexSnafu, Result, TableExistsSnafu,
+    BuildRowKeyDescriptorSnafu, InvalidPrimaryKeySnafu, MissingTimestampIndexSnafu, Result,
+    TableExistsSnafu,
 };
 use crate::table::MitoTable;
 
@@ -248,6 +249,27 @@ fn build_column_family(
     ))
 }
 
+fn validate_create_table_request(request: &CreateTableRequest) -> Result<()> {
+    let ts_index = request
+        .schema
+        .timestamp_index()
+        .context(MissingTimestampIndexSnafu {
+            table_name: &request.table_name,
+        })?;
+
+    ensure!(
+        !request
+            .primary_key_indices
+            .iter()
+            .any(|index| *index == ts_index),
+        InvalidPrimaryKeySnafu {
+            msg: "time index column can't be included in primary key"
+        }
+    );
+
+    Ok(())
+}
+
 impl<S: StorageEngine> MitoEngineInner<S> {
     async fn create_table(
         &self,
@@ -263,6 +285,8 @@ impl<S: StorageEngine> MitoEngineInner<S> {
             table: table_name,
         };
 
+        validate_create_table_request(&request)?;
+
         if let Some(table) = self.get_table(&table_ref) {
             if request.create_if_not_exists {
                 return Ok(table);
@@ -652,6 +676,49 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_validate_create_table_request() {
+        let table_name = "test_validate_create_table_request";
+        let column_schemas = vec![
+            ColumnSchema::new("name", ConcreteDataType::string_datatype(), false),
+            ColumnSchema::new(
+                "ts",
+                ConcreteDataType::timestamp_datatype(common_time::timestamp::TimeUnit::Millisecond),
+                true,
+            )
+            .with_time_index(true),
+        ];
+
+        let schema = Arc::new(
+            SchemaBuilder::try_from(column_schemas)
+                .unwrap()
+                .build()
+                .expect("ts must be timestamp column"),
+        );
+
+        let mut request = CreateTableRequest {
+            id: 1,
+            catalog_name: "greptime".to_string(),
+            schema_name: "public".to_string(),
+            table_name: table_name.to_string(),
+            desc: Some("a test table".to_string()),
+            schema,
+            create_if_not_exists: true,
+            // put ts into primary keys
+            primary_key_indices: vec![0, 1],
+            table_options: HashMap::new(),
+            region_numbers: vec![0],
+        };
+
+        let err = validate_create_table_request(&request).unwrap_err();
+        assert!(err
+            .to_string()
+            .contains("Invalid primary key: time index column can't be included in primary key"));
+
+        request.primary_key_indices = vec![0];
+        assert!(validate_create_table_request(&request).is_ok());
+    }
+
     #[tokio::test]
     async fn test_create_table_insert_scan() {
         let (_engine, table, schema, _dir) = test_util::setup_test_engine_and_table().await;

src/mito/src/error.rs

@@ -56,6 +56,9 @@ pub enum Error {
         backtrace: Backtrace,
     },
 
+    #[snafu(display("Invalid primary key: {}", msg))]
+    InvalidPrimaryKey { msg: String, backtrace: Backtrace },
+
     #[snafu(display("Missing timestamp index for table: {}", table_name))]
     MissingTimestampIndex {
         table_name: String,
@@ -214,6 +217,7 @@ impl ErrorExt for Error {
             | BuildRegionDescriptor { .. }
             | TableExists { .. }
             | ProjectedColumnNotFound { .. }
+            | InvalidPrimaryKey { .. }
             | MissingTimestampIndex { .. }
             | UnsupportedDefaultConstraint { .. }
             | TableNotFound { .. } => StatusCode::InvalidArguments,

src/script/src/table.rs

@@ -60,9 +60,9 @@ impl ScriptsTable {
             table_name: SCRIPTS_TABLE_NAME.to_string(),
             desc: Some("Scripts table".to_string()),
             schema,
-            // name and timestamp as primary key
             region_numbers: vec![0],
-            primary_key_indices: vec![0, 3],
+            // name as primary key
+            primary_key_indices: vec![0],
             create_if_not_exists: true,
             table_options: HashMap::default(),
         };

tests-integration/src/test_util.rs

@@ -193,7 +193,7 @@ pub async fn create_test_table(
                         .expect("ts is expected to be timestamp column"),
                 ),
                 create_if_not_exists: true,
-                primary_key_indices: vec![3, 0], // "host" and "ts" are primary keys
+                primary_key_indices: vec![0], // "host" is in primary keys
                 table_options: HashMap::new(),
                 region_numbers: vec![0],
             },

tests-integration/tests/grpc.rs

@@ -258,7 +258,7 @@ fn testing_create_expr() -> CreateExpr {
         desc: Some("blabla".to_string()),
         column_defs,
         time_index: "ts".to_string(),
-        primary_keys: vec!["ts".to_string(), "host".to_string()],
+        primary_keys: vec!["host".to_string()],
         create_if_not_exists: true,
         table_options: Default::default(),
         table_id: Some(MIN_USER_TABLE_ID),