ISSUE-4262: prohibits using reserved table option in create table statement.

Makefile

@@ -80,7 +80,7 @@ embedded-meta-test: build-debug
 
 stateless-test: build-debug
 	rm -rf ./_meta*/
-	ulimit -n 10000;ulimit -s 16384; bash ./scripts/ci/ci-run-stateless-tests-standalone.sh
+	ulimit -n 10000;ulimit -s 16384; bash ./scripts/ci/ci-run-tests-embedded-meta.sh
 
 management-test: build-debug
 	rm -rf ./_meta*/

query/Cargo.toml

@@ -80,6 +80,7 @@ http = "0.2.6"
 hyper = "0.14.17"
 itertools = "0.10.3"
 jwt-simple = "0.10.8"
+lazy_static = "1.4.0"
 metrics = "0.18.0"
 nom = "7.1.0"
 num = "0.4.0"

query/src/interpreters/interpreter_table_show_create.rs

@@ -26,6 +26,7 @@ use crate::catalogs::Catalog;
 use crate::interpreters::Interpreter;
 use crate::interpreters::InterpreterPtr;
 use crate::sessions::QueryContext;
+use crate::sql::is_reserved_opt_key;
 
 pub struct ShowCreateTableInterpreter {
     ctx: Arc<QueryContext>,
@@ -74,6 +75,7 @@ impl Interpreter for ShowCreateTableInterpreter {
             let mut opts = table.options().iter().collect::<Vec<_>>();
             opts.sort_by_key(|(k, _)| *k);
             opts.iter()
+                .filter(|(k, _)| !is_reserved_opt_key(k))
                 .map(|(k, v)| format!(" {}='{}'", k.to_uppercase(), v))
                 .collect::<Vec<_>>()
                 .join("")

query/src/sql/mod.rs

@@ -23,10 +23,12 @@ mod sql_common;
 mod sql_parser;
 mod sql_statement;
 pub mod statements;
+mod table_option_keys;
 
 pub use common::*;
 pub use plan_parser::PlanParser;
 pub use planner::*;
 pub use sql_common::SQLCommon;
 pub use sql_parser::DfParser;
 pub use sql_statement::*;
+pub use table_option_keys::*;

query/src/sql/statements/statement_create_table.rs

@@ -31,13 +31,14 @@ use sqlparser::ast::ObjectName;
 use super::analyzer_expr::ExpressionAnalyzer;
 use crate::catalogs::Catalog;
 use crate::sessions::QueryContext;
+use crate::sql::is_reserved_opt_key;
 use crate::sql::statements::AnalyzableStatement;
 use crate::sql::statements::AnalyzedResult;
 use crate::sql::statements::DfQueryStatement;
 use crate::sql::DfStatement;
 use crate::sql::PlanParser;
 use crate::sql::SQLCommon;
-use crate::storages::OPT_KEY_DATABASE_ID;
+use crate::sql::OPT_KEY_DATABASE_ID;
 
 #[derive(Debug, Clone, PartialEq)]
 pub struct DfCreateTable {
@@ -122,6 +123,8 @@ impl DfCreateTable {
             options: self.options.clone(),
             ..Default::default()
         };
+        self.validate_table_options()?;
+
         self.plan_with_db_id(ctx.as_ref(), db_name, meta).await
     }
 
@@ -201,4 +204,25 @@ impl DfCreateTable {
         }
         Ok(meta)
     }
+
+    fn validate_table_options(&self) -> Result<()> {
+        let reserved = self
+            .options
+            .keys()
+            .filter_map(|k| {
+                if is_reserved_opt_key(k) {
+                    Some(k.as_str())
+                } else {
+                    None
+                }
+            })
+            .collect::<Vec<_>>();
+        if !reserved.is_empty() {
+            Err(ErrorCode::BadOption(format!("the following table options are reserved, please do not specify them in the CREATE TABLE statement: {}",
+                        reserved.join(",")
+                        )))
+        } else {
+            Ok(())
+        }
+    }
 }

query/src/sql/table_option_keys.rs

@@ -0,0 +1,31 @@
+// Copyright 2021 Datafuse Labs.
+//
+// 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.
+
+use std::collections::HashSet;
+
+use lazy_static::lazy_static;
+
+pub const OPT_KEY_DATABASE_ID: &str = "database_id";
+
+lazy_static! {
+    pub static ref RESERVED_TABLE_OPTION_KEYS: HashSet<&'static str> = {
+        let mut r = HashSet::new();
+        r.insert(OPT_KEY_DATABASE_ID);
+        r
+    };
+}
+
+pub fn is_reserved_opt_key<S: AsRef<str>>(opt_key: S) -> bool {
+    RESERVED_TABLE_OPTION_KEYS.contains(opt_key.as_ref().to_lowercase().as_str())
+}

query/src/storages/fuse/constants.rs

@@ -11,14 +11,11 @@
 //  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.
-//
 
 pub const FUSE_OPT_KEY_BLOCK_IN_MEM_SIZE_THRESHOLD: &str = "block_size_threshold";
 pub const FUSE_OPT_KEY_BLOCK_PER_SEGMENT: &str = "block_per_segment";
 pub const FUSE_OPT_KEY_ROW_PER_BLOCK: &str = "row_per_block";
 pub const FUSE_OPT_KEY_SNAPSHOT_LOC: &str = "snapshot_loc";
-pub const FUSE_OPT_KEY_SNAPSHOT_SIZE: &str = "snapshot_size";
-pub const FUSE_OPT_KEY_SNAPSHOT_VER: &str = "snapshot_ver";
 
 pub const FUSE_TBL_BLOCK_PREFIX: &str = "_b";
 pub const FUSE_TBL_SEGMENT_PREFIX: &str = "_sg";

query/src/storages/fuse/fuse_history.rs

@@ -39,7 +39,7 @@ impl<'a> FuseHistory<'a> {
         let tbl = &self.table;
         let tbl_info = tbl.get_table_info();
         let snapshot_location = tbl_info.meta.options.get(FUSE_OPT_KEY_SNAPSHOT_LOC);
-        let snapshot_version = FuseTable::parse_snaphost_format_version(tbl_info)?;
+        let snapshot_version = tbl.snapshot_format_version();
         let reader = MetaReaders::table_snapshot_reader(self.ctx.as_ref());
         let snapshots = reader
             .read_snapshot_history(

query/src/storages/fuse/fuse_table.rs

@@ -32,17 +32,16 @@ use futures::StreamExt;
 
 use crate::pipelines::new::NewPipeline;
 use crate::sessions::QueryContext;
+use crate::sql::OPT_KEY_DATABASE_ID;
 use crate::storages::fuse::io::MetaReaders;
 use crate::storages::fuse::io::TableMetaLocationGenerator;
 use crate::storages::fuse::meta::TableSnapshot;
-use crate::storages::fuse::meta::DEFAULT_SNAPSHOT_VERSION;
+use crate::storages::fuse::meta::Versioned;
 use crate::storages::fuse::operations::AppendOperationLogEntry;
 use crate::storages::fuse::FUSE_OPT_KEY_SNAPSHOT_LOC;
-use crate::storages::fuse::FUSE_OPT_KEY_SNAPSHOT_VER;
 use crate::storages::StorageContext;
 use crate::storages::StorageDescription;
 use crate::storages::Table;
-use crate::storages::OPT_KEY_DATABASE_ID;
 
 #[derive(Clone)]
 pub struct FuseTable {
@@ -181,30 +180,15 @@ impl FuseTable {
             .cloned()
     }
 
-    pub fn snapshot_format_version(&self) -> Result<u64> {
-        Self::parse_snaphost_format_version(&self.table_info)
-    }
-
-    pub fn parse_snaphost_format_version(table_info: &TableInfo) -> Result<u64> {
-        table_info
-            .options()
-            .get(FUSE_OPT_KEY_SNAPSHOT_VER)
-            .map(|ver_str| {
-                let v = ver_str.as_str();
-                v.parse::<u64>().map_err(|_| {
-                    ErrorCode::LogicalError(format!(
-                        "invalid snapshot version {v}, can not be parsed as u64"
-                    ))
-                })
-            })
-            .transpose()
-            .map(|opt| {
-                opt.unwrap_or(
-                    // TableSnapshot of version other than v0 should have a explicitly define
-                    // version
-                    DEFAULT_SNAPSHOT_VERSION,
-                )
-            })
+    pub fn snapshot_format_version(&self) -> u64 {
+        match self.table_info.options().get(FUSE_OPT_KEY_SNAPSHOT_LOC) {
+            Some(loc) => TableMetaLocationGenerator::snaphost_version(loc),
+            None => {
+                // No snapshot location here, indicates that there are no data of this table yet
+                // in this case, we just returns the current snapshot version
+                TableSnapshot::VERSION
+            }
+        }
     }
 
     #[tracing::instrument(level = "debug", skip(self, ctx), fields(ctx.id = ctx.get_id().as_str()))]
@@ -214,7 +198,7 @@ impl FuseTable {
     ) -> Result<Option<Arc<TableSnapshot>>> {
         if let Some(loc) = self.snapshot_loc() {
             let reader = MetaReaders::table_snapshot_reader(ctx);
-            let ver = self.snapshot_format_version()?;
+            let ver = self.snapshot_format_version();
             Ok(Some(reader.read(loc.as_str(), None, ver).await?))
         } else {
             Ok(None)

query/src/storages/fuse/io/locations.rs

@@ -13,6 +13,8 @@
 //  limitations under the License.
 //
 
+use std::marker::PhantomData;
+
 use common_datablocks::DataBlock;
 use common_exception::Result;
 use uuid::Uuid;
@@ -22,9 +24,11 @@ use crate::storages::fuse::constants::FUSE_TBL_SEGMENT_PREFIX;
 use crate::storages::fuse::constants::FUSE_TBL_SNAPSHOT_PREFIX;
 use crate::storages::fuse::meta::SegmentInfo;
 use crate::storages::fuse::meta::SnapshotVersion;
-use crate::storages::fuse::meta::TableSnapshot;
 use crate::storages::fuse::meta::Versioned;
 
+static SNAPSHOT_V0: SnapshotVersion = SnapshotVersion::V0(PhantomData);
+static SNAPHOST_V1: SnapshotVersion = SnapshotVersion::V1(PhantomData);
+
 #[derive(Clone)]
 pub struct TableMetaLocationGenerator {
     prefix: String,
@@ -65,32 +69,36 @@ impl TableMetaLocationGenerator {
         let snaphost_version = SnapshotVersion::try_from(version)?;
         Ok(snaphost_version.create(id, &self.prefix))
     }
+
+    pub fn snaphost_version(location: impl AsRef<str>) -> u64 {
+        if location.as_ref().ends_with(SNAPHOST_V1.suffix()) {
+            SNAPHOST_V1.version()
+        } else {
+            SNAPSHOT_V0.version()
+        }
+    }
 }
 
 trait SnapshotLocationCreator {
     fn create(&self, id: &Uuid, prefix: impl AsRef<str>) -> String;
+    fn suffix(&self) -> &'static str;
 }
 
 impl SnapshotLocationCreator for SnapshotVersion {
     fn create(&self, id: &Uuid, prefix: impl AsRef<str>) -> String {
+        format!(
+            "{}/{}/{}{}",
+            prefix.as_ref(),
+            FUSE_TBL_SNAPSHOT_PREFIX,
+            id.to_simple(),
+            self.suffix(),
+        )
+    }
+
+    fn suffix(&self) -> &'static str {
         match self {
-            SnapshotVersion::V0(_) => {
-                format!(
-                    "{}/{}/{}",
-                    prefix.as_ref(),
-                    FUSE_TBL_SNAPSHOT_PREFIX,
-                    id.to_simple(),
-                )
-            }
-            SnapshotVersion::V1(_) => {
-                format!(
-                    "{}/{}/{}_v{}.json",
-                    prefix.as_ref(),
-                    FUSE_TBL_SNAPSHOT_PREFIX,
-                    id.to_simple(),
-                    TableSnapshot::VERSION,
-                )
-            }
+            SnapshotVersion::V0(_) => "",
+            SnapshotVersion::V1(_) => "_v1.json",
         }
     }
 }

query/src/storages/fuse/meta/mod.rs

@@ -19,9 +19,6 @@ mod common;
 mod current;
 mod v0;
 mod v1;
-/// where
-/// - Numeric values of version is assigned
-/// - converters from u64 into meta data version informations
 mod versions;
 
 pub use common::ColumnId;
@@ -32,4 +29,3 @@ pub use common::Versioned;
 pub use current::*;
 pub use versions::SegmentInfoVersion;
 pub use versions::SnapshotVersion;
-pub use versions::DEFAULT_SNAPSHOT_VERSION;

query/src/storages/fuse/meta/versions.rs

@@ -21,8 +21,6 @@ use crate::storages::fuse::meta::common::Versioned;
 use crate::storages::fuse::meta::v0;
 use crate::storages::fuse::meta::v1;
 
-pub const DEFAULT_SNAPSHOT_VERSION: u64 = v0::TableSnapshot::VERSION;
-
 // Here versions of meta are tagged with numeric values
 //
 // The trait Versioned itself can not prevent us from
@@ -51,6 +49,19 @@ pub enum SnapshotVersion {
     V1(PhantomData<v1::TableSnapshot>),
 }
 
+impl SnapshotVersion {
+    pub fn version(&self) -> u64 {
+        match self {
+            SnapshotVersion::V0(a) => Self::ver(a),
+            SnapshotVersion::V1(a) => Self::ver(a),
+        }
+    }
+
+    fn ver<const V: u64, T: Versioned<V>>(_v: &PhantomData<T>) -> u64 {
+        V
+    }
+}
+
 impl Versioned<0> for DataBlock {}
 
 #[allow(dead_code)]
@@ -94,8 +105,8 @@ mod converters {
 
     /// Statically check that if T implements Versoined<U> where U equals V
     #[inline]
-    fn ver_eq<T, const V: u64>(v: PhantomData<T>) -> PhantomData<T>
+    fn ver_eq<T, const V: u64>(t: PhantomData<T>) -> PhantomData<T>
     where T: Versioned<V> {
-        v
+        t
     }
 }

query/src/storages/fuse/operations/commit.rs

@@ -44,7 +44,6 @@ use crate::storages::fuse::operations::TableOperationLog;
 use crate::storages::fuse::statistics;
 use crate::storages::fuse::FuseTable;
 use crate::storages::fuse::FUSE_OPT_KEY_SNAPSHOT_LOC;
-use crate::storages::fuse::FUSE_OPT_KEY_SNAPSHOT_VER;
 use crate::storages::Table;
 
 impl FuseTable {
@@ -141,7 +140,7 @@ impl FuseTable {
         overwrite: bool,
     ) -> Result<()> {
         let prev = self.read_table_snapshot(ctx).await?;
-        let prev_version = self.snapshot_format_version()?;
+        let prev_version = self.snapshot_format_version();
         let schema = self.table_info.meta.schema.as_ref().clone();
         let (segments, summary) = Self::merge_append_operations(&schema, operation_log)?;
 
@@ -240,16 +239,10 @@ impl FuseTable {
         let req = UpsertTableOptionReq {
             table_id,
             seq: MatchSeq::Exact(tbl_id.version),
-            options: [
-                (
-                    FUSE_OPT_KEY_SNAPSHOT_LOC.to_owned(),
-                    Some(new_snapshot_location),
-                ),
-                (
-                    FUSE_OPT_KEY_SNAPSHOT_VER.to_owned(),
-                    Some(TableSnapshot::VERSION.to_string()),
-                ),
-            ]
+            options: [(
+                FUSE_OPT_KEY_SNAPSHOT_LOC.to_owned(),
+                Some(new_snapshot_location),
+            )]
             .into_iter()
             .collect(),
         };