fix: allow passing extra table options (#3484)

* fix: do not check options in parser

* test: fix tests

* test: fix sqlness

* test: add sqlness test

* chore: log options

* chore: must specify compaction type

* feat: validate option key

* feat: add option key validation back

src/common/datasource/src/object_store/s3.rs

@@ -28,12 +28,15 @@ const REGION: &str = "region";
 const ENABLE_VIRTUAL_HOST_STYLE: &str = "enable_virtual_host_style";
 
 pub fn is_supported_in_s3(key: &str) -> bool {
-    key == ENDPOINT
-        || key == ACCESS_KEY_ID
-        || key == SECRET_ACCESS_KEY
-        || key == SESSION_TOKEN
-        || key == REGION
-        || key == ENABLE_VIRTUAL_HOST_STYLE
+    [
+        ENDPOINT,
+        ACCESS_KEY_ID,
+        SECRET_ACCESS_KEY,
+        SESSION_TOKEN,
+        REGION,
+        ENABLE_VIRTUAL_HOST_STYLE,
+    ]
+    .contains(&key)
 }
 
 pub fn build_s3_backend(

src/mito2/src/region/opener.rs

@@ -171,6 +171,8 @@ impl RegionOpener {
         // Initial memtable id is 0.
         let mutable = self.memtable_builder.build(0, &metadata);
 
+        debug!("Create region {} with options: {:?}", region_id, options);
+
         let version = VersionBuilder::new(metadata, mutable)
             .options(options)
             .build();
@@ -249,6 +251,9 @@ impl RegionOpener {
 
         let region_id = self.region_id;
         let object_store = self.object_store(&region_options.storage)?.clone();
+
+        debug!("Open region {} with options: {:?}", region_id, self.options);
+
         let access_layer = Arc::new(AccessLayer::new(
             self.region_dir.clone(),
             object_store,

src/mito2/src/region/options.rs

@@ -13,6 +13,8 @@
 // limitations under the License.
 
 //! Options for a region.
+//!
+//! If we add options in this mod, we also need to modify [store_api::mito_engine_options].
 
 use std::collections::HashMap;
 use std::time::Duration;
@@ -358,6 +360,7 @@ mod tests {
             ("compaction.type", "twcs"),
             ("storage", "S3"),
             ("index.inverted_index.ignore_column_ids", "1,2,3"),
+            ("index.inverted_index.segment_row_count", "512"),
             (
                 WAL_OPTIONS_KEY,
                 &serde_json::to_string(&wal_options).unwrap(),
@@ -376,7 +379,7 @@ mod tests {
             index_options: IndexOptions {
                 inverted_index: InvertedIndexOptions {
                     ignore_column_ids: vec![1, 2, 3],
-                    segment_row_count: 1024,
+                    segment_row_count: 512,
                 },
             },
         };

src/sql/src/error.rs

@@ -117,7 +117,7 @@ pub enum Error {
         source: datatypes::error::Error,
     },
 
-    #[snafu(display("Invalid table option key: {}", key))]
+    #[snafu(display("Unrecognized table option key: {}", key))]
     InvalidTableOption { key: String, location: Location },
 
     #[snafu(display("Failed to serialize column default constraint"))]

src/sql/src/parsers/create_parser.rs

@@ -23,7 +23,7 @@ use sqlparser::keywords::ALL_KEYWORDS;
 use sqlparser::parser::IsOptional::Mandatory;
 use sqlparser::parser::{Parser, ParserError};
 use sqlparser::tokenizer::{Token, TokenWithLocation, Word};
-use table::requests::valid_table_option;
+use table::requests::validate_table_option;
 
 use crate::ast::{ColumnDef, Ident, TableConstraint};
 use crate::error::{
@@ -84,7 +84,7 @@ impl<'a> ParserContext<'a> {
             .collect::<HashMap<String, String>>();
         for key in options.keys() {
             ensure!(
-                valid_table_option(key),
+                validate_table_option(key),
                 InvalidTableOptionSnafu {
                     key: key.to_string()
                 }
@@ -150,7 +150,7 @@ impl<'a> ParserContext<'a> {
             .context(error::SyntaxSnafu)?;
         for option in options.iter() {
             ensure!(
-                valid_table_option(&option.name.value),
+                validate_table_option(&option.name.value),
                 InvalidTableOptionSnafu {
                     key: option.name.value.to_string()
                 }
@@ -799,6 +799,17 @@ mod tests {
                 expected_engine: "foo",
                 expected_if_not_exist: true,
             },
+            Test {
+                sql: "CREATE EXTERNAL TABLE IF NOT EXISTS city ENGINE=foo with(location='/var/data/city.csv',format='csv','compaction.type'='bar');",
+                expected_table_name: "city",
+                expected_options: HashMap::from([
+                    ("location".to_string(), "/var/data/city.csv".to_string()),
+                    ("format".to_string(), "csv".to_string()),
+                    ("compaction.type".to_string(), "bar".to_string()),
+                ]),
+                expected_engine: "foo",
+                expected_if_not_exist: true,
+            },
         ];
 
         for test in tests {

src/sql/src/statements/create.rs

@@ -230,7 +230,7 @@ pub struct CreateTableLike {
 #[cfg(test)]
 mod tests {
     use crate::dialect::GreptimeDbDialect;
-    use crate::error::Error::InvalidTableOption;
+    use crate::error::Error;
     use crate::parser::{ParseOptions, ParserContext};
     use crate::statements::statement::Statement;
 
@@ -344,7 +344,29 @@ ENGINE=mito
     fn test_validate_table_options() {
         let sql = r"create table if not exists demo(
             host string,
-            ts bigint,
+            ts timestamp,
+            cpu double default 0,
+            memory double,
+            TIME INDEX (ts),
+            PRIMARY KEY(host)
+      )
+      PARTITION ON COLUMNS (host) ()
+      engine=mito
+      with(regions=1, ttl='7d', 'compaction.type'='world');
+";
+        let result =
+            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
+                .unwrap();
+        match &result[0] {
+            Statement::CreateTable(c) => {
+                assert_eq!(3, c.options.len());
+            }
+            _ => unreachable!(),
+        }
+
+        let sql = r"create table if not exists demo(
+            host string,
+            ts timestamp,
             cpu double default 0,
             memory double,
             TIME INDEX (ts),
@@ -356,6 +378,6 @@ ENGINE=mito
 ";
         let result =
             ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
-        assert!(matches!(result, Err(InvalidTableOption { .. })))
+        assert!(matches!(result, Err(Error::InvalidTableOption { .. })));
     }
 }

src/store-api/src/lib.rs

@@ -20,6 +20,7 @@ pub mod logstore;
 pub mod manifest;
 pub mod metadata;
 pub mod metric_engine_consts;
+pub mod mito_engine_options;
 pub mod path_utils;
 pub mod region_engine;
 pub mod region_request;

src/store-api/src/mito_engine_options.rs

@@ -0,0 +1,61 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Option keys for the mito engine.
+//! We define them in this mod so the create parser can use it to validate table options.
+
+use common_wal::options::WAL_OPTIONS_KEY;
+
+/// Returns true if the `key` is a valid option key for the mito engine.
+pub fn is_mito_engine_option_key(key: &str) -> bool {
+    [
+        "ttl",
+        "compaction.type",
+        "compaction.twcs.max_active_window_files",
+        "compaction.twcs.max_inactive_window_files",
+        "compaction.twcs.time_window",
+        "storage",
+        "index.inverted_index.ignore_column_ids",
+        "index.inverted_index.segment_row_count",
+        WAL_OPTIONS_KEY,
+    ]
+    .contains(&key)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_is_mito_engine_option_key() {
+        assert!(is_mito_engine_option_key("ttl"));
+        assert!(is_mito_engine_option_key("compaction.type"));
+        assert!(is_mito_engine_option_key(
+            "compaction.twcs.max_active_window_files"
+        ));
+        assert!(is_mito_engine_option_key(
+            "compaction.twcs.max_inactive_window_files"
+        ));
+        assert!(is_mito_engine_option_key("compaction.twcs.time_window"));
+        assert!(is_mito_engine_option_key("storage"));
+        assert!(is_mito_engine_option_key(
+            "index.inverted_index.ignore_column_ids"
+        ));
+        assert!(is_mito_engine_option_key(
+            "index.inverted_index.segment_row_count"
+        ));
+        assert!(is_mito_engine_option_key("wal_options"));
+        assert!(!is_mito_engine_option_key("foo"));
+    }
+}

src/table/src/requests.rs

@@ -26,6 +26,7 @@ use datatypes::prelude::VectorRef;
 use datatypes::schema::{ColumnSchema, RawSchema};
 use serde::{Deserialize, Serialize};
 use store_api::metric_engine_consts::{LOGICAL_TABLE_METADATA_KEY, PHYSICAL_TABLE_METADATA_KEY};
+use store_api::mito_engine_options::is_mito_engine_option_key;
 use store_api::storage::RegionNumber;
 
 use crate::error;
@@ -38,6 +39,33 @@ pub const FILE_TABLE_LOCATION_KEY: &str = "location";
 pub const FILE_TABLE_PATTERN_KEY: &str = "pattern";
 pub const FILE_TABLE_FORMAT_KEY: &str = "format";
 
+/// Returns true if the `key` is a valid key for any engine or storage.
+pub fn validate_table_option(key: &str) -> bool {
+    if is_supported_in_s3(key) {
+        return true;
+    }
+
+    if is_mito_engine_option_key(key) {
+        return true;
+    }
+
+    [
+        // common keys:
+        WRITE_BUFFER_SIZE_KEY,
+        TTL_KEY,
+        REGIONS_KEY,
+        STORAGE_KEY,
+        // file engine keys:
+        FILE_TABLE_LOCATION_KEY,
+        FILE_TABLE_FORMAT_KEY,
+        FILE_TABLE_PATTERN_KEY,
+        // metric engine keys:
+        PHYSICAL_TABLE_METADATA_KEY,
+        LOGICAL_TABLE_METADATA_KEY,
+    ]
+    .contains(&key)
+}
+
 #[derive(Debug, Clone)]
 pub struct CreateDatabaseRequest {
     pub db_name: String,
@@ -315,21 +343,6 @@ impl TruncateTableRequest {
     }
 }
 
-pub fn valid_table_option(key: &str) -> bool {
-    matches!(
-        key,
-        FILE_TABLE_LOCATION_KEY
-            | FILE_TABLE_FORMAT_KEY
-            | FILE_TABLE_PATTERN_KEY
-            | WRITE_BUFFER_SIZE_KEY
-            | TTL_KEY
-            | REGIONS_KEY
-            | STORAGE_KEY
-            | PHYSICAL_TABLE_METADATA_KEY
-            | LOGICAL_TABLE_METADATA_KEY
-    ) | is_supported_in_s3(key)
-}
-
 #[derive(Debug, Clone, Default, Deserialize, Serialize)]
 pub struct CopyDatabaseRequest {
     pub catalog_name: String,
@@ -346,14 +359,14 @@ mod tests {
 
     #[test]
     fn test_validate_table_option() {
-        assert!(valid_table_option(FILE_TABLE_LOCATION_KEY));
-        assert!(valid_table_option(FILE_TABLE_FORMAT_KEY));
-        assert!(valid_table_option(FILE_TABLE_PATTERN_KEY));
-        assert!(valid_table_option(TTL_KEY));
-        assert!(valid_table_option(REGIONS_KEY));
-        assert!(valid_table_option(WRITE_BUFFER_SIZE_KEY));
-        assert!(valid_table_option(STORAGE_KEY));
-        assert!(!valid_table_option("foo"));
+        assert!(validate_table_option(FILE_TABLE_LOCATION_KEY));
+        assert!(validate_table_option(FILE_TABLE_FORMAT_KEY));
+        assert!(validate_table_option(FILE_TABLE_PATTERN_KEY));
+        assert!(validate_table_option(TTL_KEY));
+        assert!(validate_table_option(REGIONS_KEY));
+        assert!(validate_table_option(WRITE_BUFFER_SIZE_KEY));
+        assert!(validate_table_option(STORAGE_KEY));
+        assert!(!validate_table_option("foo"));
     }
 
     #[test]

tests-integration/src/tests/instance_test.rs

@@ -160,7 +160,7 @@ PARTITION ON COLUMNS (n) (
 }
 
 #[apply(both_instances_cases)]
-async fn test_validate_external_table_options(instance: Arc<dyn MockInstance>) {
+async fn test_extra_external_table_options(instance: Arc<dyn MockInstance>) {
     let frontend = instance.frontend();
     let format = "json";
     let location = find_testing_resource("/tests/data/json/various_type.json");