feat(query): create table support add inverted index (#15547)

Author: b41sh

src/meta/api/src/schema_api_impl.rs

@@ -1557,6 +1557,26 @@ impl<KV: kvapi::KVApi<Error = MetaError> + ?Sized> SchemaApi for KV {
                 .collect::<Vec<_>>()
         };
 
+        if !req.table_meta.indexes.is_empty() {
+            // check the index column id exists and not be duplicated.
+            let mut index_column_ids = HashSet::new();
+            for (_, index) in req.table_meta.indexes.iter() {
+                for column_id in &index.column_ids {
+                    if req.table_meta.schema.is_column_deleted(*column_id) {
+                        return Err(KVAppError::AppError(AppError::IndexColumnIdNotFound(
+                            IndexColumnIdNotFound::new(*column_id, &index.name),
+                        )));
+                    }
+                    if index_column_ids.contains(column_id) {
+                        return Err(KVAppError::AppError(AppError::DuplicatedIndexColumnId(
+                            DuplicatedIndexColumnId::new(*column_id, &index.name),
+                        )));
+                    }
+                    index_column_ids.insert(column_id);
+                }
+            }
+        }
+
         let mut trials = txn_backoff(None, func_name!());
         loop {
             trials.next().unwrap()?.await;

src/query/ast/src/ast/format/ast_format.rs

@@ -1470,12 +1470,18 @@ impl<'ast> Visitor<'ast> for AstFormatVisitor {
 
     fn visit_create_table_source(&mut self, source: &'ast CreateTableSource) {
         match source {
-            CreateTableSource::Columns(columns) => {
+            CreateTableSource::Columns(columns, inverted_indexes) => {
                 let mut children = Vec::with_capacity(columns.len());
                 for column in columns.iter() {
                     self.visit_column_definition(column);
                     children.push(self.children.pop().unwrap());
                 }
+                if let Some(inverted_indexes) = inverted_indexes {
+                    for inverted_index in inverted_indexes {
+                        self.visit_inverted_index_definition(inverted_index);
+                        children.push(self.children.pop().unwrap());
+                    }
+                }
                 let name = "ColumnsDefinition".to_string();
                 let format_ctx = AstFormatContext::with_children(name, children.len());
                 let node = FormatTreeNode::with_children(format_ctx, children);
@@ -1507,6 +1513,26 @@ impl<'ast> Visitor<'ast> for AstFormatVisitor {
         self.children.push(node);
     }
 
+    fn visit_inverted_index_definition(
+        &mut self,
+        inverted_index_definition: &'ast InvertedIndexDefinition,
+    ) {
+        let mut column_nodes = Vec::with_capacity(inverted_index_definition.columns.len());
+        for column in &inverted_index_definition.columns {
+            let column_name = format!("Column {}", column);
+            let column_format_ctx = AstFormatContext::new(column_name);
+            let column_node = FormatTreeNode::new(column_format_ctx);
+            column_nodes.push(column_node);
+        }
+        let name = format!(
+            "InvertedIndexDefinition {}",
+            inverted_index_definition.index_name
+        );
+        let format_ctx = AstFormatContext::with_children(name, column_nodes.len());
+        let node = FormatTreeNode::with_children(format_ctx, column_nodes);
+        self.children.push(node);
+    }
+
     fn visit_drop_table(&mut self, stmt: &'ast DropTableStmt) {
         self.visit_table_ref(&stmt.catalog, &stmt.database, &stmt.table);
         let child = self.children.pop().unwrap();

src/query/ast/src/ast/format/syntax/ddl.rs

@@ -113,14 +113,27 @@ pub(crate) fn pretty_create_table(stmt: CreateTableStmt) -> RcDoc<'static> {
 
 fn pretty_table_source(source: CreateTableSource) -> RcDoc<'static> {
     match source {
-        CreateTableSource::Columns(columns) => RcDoc::space().append(parenthesized(
-            interweave_comma(
-                columns
-                    .into_iter()
-                    .map(|column| RcDoc::text(column.to_string())),
-            )
-            .group(),
-        )),
+        CreateTableSource::Columns(columns, inverted_indexes) => RcDoc::space()
+            .append(parenthesized(
+                interweave_comma(
+                    columns
+                        .into_iter()
+                        .map(|column| RcDoc::text(column.to_string())),
+                )
+                .group(),
+            ))
+            .append(if let Some(inverted_indexes) = inverted_indexes {
+                parenthesized(
+                    interweave_comma(
+                        inverted_indexes
+                            .into_iter()
+                            .map(|inverted_index| RcDoc::text(inverted_index.to_string())),
+                    )
+                    .group(),
+                )
+            } else {
+                RcDoc::nil()
+            }),
         CreateTableSource::Like {
             catalog,
             database,

src/query/ast/src/ast/statements/table.rs

@@ -233,7 +233,7 @@ impl Display for AttachTableStmt {
 
 #[derive(Debug, Clone, PartialEq, Drive, DriveMut)]
 pub enum CreateTableSource {
-    Columns(Vec<ColumnDefinition>),
+    Columns(Vec<ColumnDefinition>, Option<Vec<InvertedIndexDefinition>>),
     Like {
         catalog: Option<Identifier>,
         database: Option<Identifier>,
@@ -244,9 +244,13 @@ pub enum CreateTableSource {
 impl Display for CreateTableSource {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         match self {
-            CreateTableSource::Columns(columns) => {
+            CreateTableSource::Columns(columns, inverted_indexes) => {
                 write!(f, "(")?;
                 write_comma_separated_list(f, columns)?;
+                if let Some(inverted_indexes) = inverted_indexes {
+                    write!(f, ", ")?;
+                    write_comma_separated_list(f, inverted_indexes)?;
+                }
                 write!(f, ")")
             }
             CreateTableSource::Like {
@@ -834,6 +838,56 @@ impl Display for ColumnDefinition {
     }
 }
 
+#[derive(Debug, Clone, PartialEq, Drive, DriveMut)]
+pub struct InvertedIndexDefinition {
+    pub index_name: Identifier,
+    pub columns: Vec<Identifier>,
+    #[drive(skip)]
+    pub sync_creation: bool,
+    #[drive(skip)]
+    pub index_options: BTreeMap<String, String>,
+}
+
+impl Display for InvertedIndexDefinition {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        if !self.sync_creation {
+            write!(f, "ASYNC ")?;
+        }
+        write!(f, "INVERTED INDEX")?;
+        write!(f, " {}", self.index_name)?;
+        write!(f, " (")?;
+        write_comma_separated_list(f, &self.columns)?;
+        write!(f, ")")?;
+
+        if !self.index_options.is_empty() {
+            write!(f, " ")?;
+            write_space_separated_string_map(f, &self.index_options)?;
+        }
+
+        Ok(())
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Drive, DriveMut)]
+pub enum CreateDefinition {
+    Column(ColumnDefinition),
+    InvertedIndex(InvertedIndexDefinition),
+}
+
+impl Display for CreateDefinition {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        match self {
+            CreateDefinition::Column(column_def) => {
+                write!(f, "{}", column_def)?;
+            }
+            CreateDefinition::InvertedIndex(inverted_index_def) => {
+                write!(f, "{}", inverted_index_def)?;
+            }
+        }
+        Ok(())
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Drive, DriveMut)]
 pub enum ModifyColumnAction {
     // (column name id, masking policy name)

src/query/ast/src/ast/visitors/visitor.rs

@@ -528,6 +528,12 @@ pub trait Visitor<'ast>: Sized {
 
     fn visit_column_definition(&mut self, _column_definition: &'ast ColumnDefinition) {}
 
+    fn visit_inverted_index_definition(
+        &mut self,
+        _inverted_index_definition: &'ast InvertedIndexDefinition,
+    ) {
+    }
+
     fn visit_drop_table(&mut self, _stmt: &'ast DropTableStmt) {}
 
     fn visit_undrop_table(&mut self, _stmt: &'ast UndropTableStmt) {}

src/query/ast/src/ast/visitors/visitor_mut.rs

@@ -540,6 +540,12 @@ pub trait VisitorMut: Sized {
 
     fn visit_column_definition(&mut self, _column_definition: &mut ColumnDefinition) {}
 
+    fn visit_inverted_index_definition(
+        &mut self,
+        _inverted_index_definition: &mut InvertedIndexDefinition,
+    ) {
+    }
+
     fn visit_drop_table(&mut self, _stmt: &mut DropTableStmt) {}
 
     fn visit_undrop_table(&mut self, _stmt: &mut UndropTableStmt) {}

src/query/ast/src/parser/statement.rs

@@ -2725,6 +2725,36 @@ pub fn column_def(i: Input) -> IResult<ColumnDefinition> {
     Ok((i, def))
 }
 
+pub fn inverted_index_def(i: Input) -> IResult<InvertedIndexDefinition> {
+    map_res(
+        rule! {
+            ASYNC?
+            ~ INVERTED ~ ^INDEX
+            ~ #ident
+            ~ ^"(" ~ ^#comma_separated_list1(ident) ~ ^")"
+            ~ ( #table_option )?
+        },
+        |(opt_async, _, _, index_name, _, columns, _, opt_index_options)| {
+            Ok(InvertedIndexDefinition {
+                index_name,
+                columns,
+                sync_creation: opt_async.is_none(),
+                index_options: opt_index_options.unwrap_or_default(),
+            })
+        },
+    )(i)
+}
+
+pub fn create_def(i: Input) -> IResult<CreateDefinition> {
+    alt((
+        map(rule! { #column_def }, CreateDefinition::Column),
+        map(
+            rule! { #inverted_index_def },
+            CreateDefinition::InvertedIndex,
+        ),
+    ))(i)
+}
+
 pub fn role_name(i: Input) -> IResult<String> {
     let role_ident = map(
         rule! {
@@ -3036,9 +3066,28 @@ pub fn grant_option(i: Input) -> IResult<PrincipalIdentity> {
 pub fn create_table_source(i: Input) -> IResult<CreateTableSource> {
     let columns = map(
         rule! {
-            "(" ~ ^#comma_separated_list1(column_def) ~ ^")"
+            "(" ~ ^#comma_separated_list1(create_def) ~ ^")"
+        },
+        |(_, create_defs, _)| {
+            let mut columns = Vec::with_capacity(create_defs.len());
+            let mut inverted_indexes = Vec::new();
+            for create_def in create_defs {
+                match create_def {
+                    CreateDefinition::Column(column) => {
+                        columns.push(column);
+                    }
+                    CreateDefinition::InvertedIndex(inverted_index) => {
+                        inverted_indexes.push(inverted_index);
+                    }
+                }
+            }
+            let opt_inverted_indexes = if !inverted_indexes.is_empty() {
+                Some(inverted_indexes)
+            } else {
+                None
+            };
+            CreateTableSource::Columns(columns, opt_inverted_indexes)
         },
-        |(_, columns, _)| CreateTableSource::Columns(columns),
     );
     let like = map(
         rule! {

src/query/ast/tests/it/parser.rs

@@ -129,6 +129,7 @@ fn test_statement() {
         r#"create table if not exists a.b (c tuple(m integer, n string), d tuple(integer, string));"#,
         r#"create table if not exists a.b (a string, b string, c string as (concat(a, ' ', b)) stored );"#,
         r#"create table if not exists a.b (a int, b int, c int generated always as (a + b) virtual );"#,
+        r#"create table if not exists a.b (a string, b string, inverted index idx1 (a,b) tokenizer='chinese');"#,
         r#"create table a.b like c.d;"#,
         r#"create table t like t2 engine = memory;"#,
         r#"create table if not exists a.b (a int) 's3://testbucket/admin/data/' connection=(aws_key_id='minioadmin' aws_secret_key='minioadmin' endpoint_url='http://127.0.0.1:9900');"#,