feat: reserve internal column (#2396)

* feat: reserve internal column

Signed-off-by: ZhuZiyi <zyzhu2001@gmail.com>

* test: add function test

Signed-off-by: ZhuZiyi <zyzhu2001@gmail.com>

* fix: spell typos

Signed-off-by: ZhuZiyi <zyzhu2001@gmail.com>

* chore: resolve conversation

Signed-off-by: ZhuZiyi <zyzhu2001@gmail.com>

---------

Signed-off-by: ZhuZiyi <zyzhu2001@gmail.com>

src/store-api/src/metadata.rs

@@ -33,6 +33,7 @@ use serde::{Deserialize, Deserializer, Serialize};
 use snafu::{ensure, Location, OptionExt, ResultExt, Snafu};
 
 use crate::region_request::{AddColumn, AddColumnLocation, AlterKind};
+use crate::storage::consts::is_internal_column;
 use crate::storage::{ColumnId, RegionId};
 
 pub type Result<T> = std::result::Result<T, MetadataError>;
@@ -368,6 +369,16 @@ impl RegionMetadata {
             );
         }
 
+        ensure!(
+            !is_internal_column(&column_metadata.column_schema.name),
+            InvalidMetaSnafu {
+                reason: format!(
+                    "{} is internal column name that can not be used",
+                    column_metadata.column_schema.name
+                ),
+            }
+        );
+
         Ok(())
     }
 }
@@ -998,4 +1009,24 @@ mod test {
         let err = builder.build().unwrap_err();
         assert_eq!(StatusCode::InvalidArguments, err.status_code());
     }
+
+    #[test]
+    fn test_invalid_column_name() {
+        let mut builder = create_builder();
+        builder.push_column_metadata(ColumnMetadata {
+            column_schema: ColumnSchema::new(
+                "__sequence",
+                ConcreteDataType::timestamp_millisecond_datatype(),
+                false,
+            ),
+            semantic_type: SemanticType::Timestamp,
+            column_id: 1,
+        });
+        let err = builder.build().unwrap_err();
+        assert!(
+            err.to_string()
+                .contains("internal column name that can not be used"),
+            "unexpected err: {err}",
+        );
+    }
 }

src/store-api/src/storage/consts.rs

@@ -84,6 +84,17 @@ pub const OP_TYPE_COLUMN_NAME: &str = "__op_type";
 /// Name for reserved column: primary_key
 pub const PRIMARY_KEY_COLUMN_NAME: &str = "__primary_key";
 
+/// Internal Column Name
+static INTERNAL_COLUMN_VEC: [&str; 3] = [
+    SEQUENCE_COLUMN_NAME,
+    OP_TYPE_COLUMN_NAME,
+    PRIMARY_KEY_COLUMN_NAME,
+];
+
+pub fn is_internal_column(name: &str) -> bool {
+    INTERNAL_COLUMN_VEC.contains(&name)
+}
+
 // -----------------------------------------------------------------------------
 
 // ---------- Default options --------------------------------------------------
@@ -104,4 +115,18 @@ mod tests {
         assert_eq!(0x80000001, ReservedColumnId::sequence());
         assert_eq!(0x80000002, ReservedColumnId::op_type());
     }
+
+    #[test]
+    fn test_is_internal_column() {
+        // contain internal column names
+        assert!(is_internal_column(SEQUENCE_COLUMN_NAME));
+        assert!(is_internal_column(OP_TYPE_COLUMN_NAME));
+        assert!(is_internal_column(PRIMARY_KEY_COLUMN_NAME));
+
+        // don't contain internal column names
+        assert!(!is_internal_column("my__column"));
+        assert!(!is_internal_column("my__sequence"));
+        assert!(!is_internal_column("my__op_type"));
+        assert!(!is_internal_column("my__primary_key"));
+    }
 }