feat(query): Support CREATE/DROP SPATIAL INDEX (Geometry/Geography types)

src/meta/api/src/schema_api.rs

@@ -774,6 +774,7 @@ pub fn mark_table_index_as_deleted(
         TableIndexType::Inverted => MarkedDeletedIndexType::INVERTED,
         TableIndexType::Ngram => MarkedDeletedIndexType::NGRAM,
         TableIndexType::Vector => MarkedDeletedIndexType::VECTOR,
+        TableIndexType::Spatial => MarkedDeletedIndexType::SPATIAL,
     };
     let marked_deleted_table_index_meta = MarkedDeletedIndexMeta {
         dropped_on: Utc::now(),

src/meta/app/src/schema/index.rs

@@ -74,6 +74,7 @@ pub enum MarkedDeletedIndexType {
     INVERTED = 2,
     NGRAM = 3,
     VECTOR = 4,
+    SPATIAL = 5,
 }
 
 #[derive(Clone, Debug, Eq, PartialEq)]

src/meta/app/src/schema/table/mod.rs

@@ -286,6 +286,7 @@ pub enum TableIndexType {
     Inverted = 0,
     Ngram = 1,
     Vector = 2,
+    Spatial = 3,
 }
 
 #[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Eq, PartialEq)]
@@ -523,6 +524,9 @@ impl Display for TableIndexType {
             TableIndexType::Vector => {
                 write!(f, "VECTOR")
             }
+            TableIndexType::Spatial => {
+                write!(f, "SPATIAL")
+            }
         }
     }
 }

src/query/ast/src/ast/statements/index.rs

@@ -44,6 +44,7 @@ pub enum TableIndexType {
     Inverted,
     Ngram,
     Vector,
+    Spatial,
 }
 
 impl Display for TableIndexType {
@@ -61,6 +62,9 @@ impl Display for TableIndexType {
             TableIndexType::Vector => {
                 write!(f, "VECTOR")
             }
+            TableIndexType::Spatial => {
+                write!(f, "SPATIAL")
+            }
         }
     }
 }

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

@@ -6367,6 +6367,7 @@ fn index_type(i: Input) -> IResult<TableIndexType> {
         value(TableIndexType::Inverted, rule! { INVERTED }),
         value(TableIndexType::Ngram, rule! { NGRAM }),
         value(TableIndexType::Vector, rule! { VECTOR }),
+        value(TableIndexType::Spatial, rule! { SPATIAL }),
     ))
     .parse(i)
 }

src/query/ast/src/parser/token.rs

@@ -1139,6 +1139,8 @@ pub enum TokenKind {
     SESSION,
     #[token("SETTINGS", ignore(ascii_case))]
     SETTINGS,
+    #[token("SPATIAL", ignore(ascii_case))]
+    SPATIAL,
     #[token("VARIABLES", ignore(ascii_case))]
     VARIABLES,
     #[token("STAGES", ignore(ascii_case))]

src/query/ast/tests/it/testdata/stmt-error.txt

@@ -217,7 +217,7 @@ error:
   --> SQL:1:6
   |
 1 | drop a
-  |      ^ unexpected `a`, expecting `TASK`, `TABLE`, `CATALOG`, `MASKING`, `DATABASE`, `PASSWORD`, `WAREHOUSE`, `AGGREGATING`, `WORKLOAD`, `SCHEMA`, `DICTIONARY`, `VIEW`, `INVERTED`, `NGRAM`, `VECTOR`, `STAGE`, `TAG`, `FILE`, `PIPE`, `NOTIFICATION`, `CONNECTION`, `ROW`, `USER`, `ROLE`, `FUNCTION`, `NETWORK`, `PROCEDURE`, `STREAM`, or `SEQUENCE`
+  |      ^ unexpected `a`, expecting `TASK`, `TABLE`, `CATALOG`, `SPATIAL`, `MASKING`, `DATABASE`, `PASSWORD`, `WAREHOUSE`, `AGGREGATING`, `WORKLOAD`, `SCHEMA`, `DICTIONARY`, `VIEW`, `INVERTED`, `NGRAM`, `VECTOR`, `STAGE`, `TAG`, `FILE`, `PIPE`, `NOTIFICATION`, `CONNECTION`, `ROW`, `USER`, `ROLE`, `FUNCTION`, `NETWORK`, `PROCEDURE`, `STREAM`, or `SEQUENCE`
 
 
 ---------- Input ----------
@@ -281,7 +281,7 @@ error:
   --> SQL:1:6
   |
 1 | drop usar if exists 'test-j';
-  |      ^^^^ unexpected `usar`, expecting `USER`, `STREAM`, `STAGE`, `PASSWORD`, `WAREHOUSE`, `AGGREGATING`, `TAG`, `ROLE`, `TABLE`, `SCHEMA`, `VECTOR`, `CATALOG`, `NETWORK`, `WORKLOAD`, `DATABASE`, `INVERTED`, `FUNCTION`, `PROCEDURE`, `TASK`, `NOTIFICATION`, `NGRAM`, `MASKING`, `SEQUENCE`, `DICTIONARY`, `VIEW`, `FILE`, `PIPE`, `CONNECTION`, or `ROW`
+  |      ^^^^ unexpected `usar`, expecting `USER`, `STREAM`, `STAGE`, `SPATIAL`, `PASSWORD`, `WAREHOUSE`, `AGGREGATING`, `TAG`, `ROLE`, `TABLE`, `SCHEMA`, `VECTOR`, `CATALOG`, `NETWORK`, `WORKLOAD`, `DATABASE`, `INVERTED`, `FUNCTION`, `PROCEDURE`, `TASK`, `NOTIFICATION`, `NGRAM`, `MASKING`, `SEQUENCE`, `DICTIONARY`, `VIEW`, `FILE`, `PIPE`, `CONNECTION`, or `ROW`
 
 
 ---------- Input ----------

src/query/service/src/interpreters/hook/refresh_hook.rs

@@ -101,8 +101,7 @@ async fn do_refresh(ctx: Arc<QueryContext>, desc: RefreshDesc) -> Result<()> {
     }
 
     // Generate sync inverted indexes.
-    let inverted_index_plans =
-        generate_refresh_inverted_index_plan(ctx.clone(), &desc, table).await?;
+    let inverted_index_plans = generate_refresh_table_index_plan(ctx.clone(), &desc, table).await?;
     plans.extend_from_slice(&inverted_index_plans);
 
     let mut tasks = Vec::with_capacity(std::cmp::min(
@@ -257,7 +256,7 @@ async fn build_refresh_index_plan(
         .await
 }
 
-async fn generate_refresh_inverted_index_plan(
+async fn generate_refresh_table_index_plan(
     ctx: Arc<QueryContext>,
     desc: &RefreshDesc,
     table: Arc<dyn Table>,
@@ -274,6 +273,7 @@ async fn generate_refresh_inverted_index_plan(
             TableIndexType::Inverted => ast::TableIndexType::Inverted,
             TableIndexType::Ngram => ast::TableIndexType::Ngram,
             TableIndexType::Vector => ast::TableIndexType::Vector,
+            TableIndexType::Spatial => ast::TableIndexType::Spatial,
         };
         let plan = RefreshTableIndexPlan {
             index_type,

src/query/service/src/interpreters/interpreter_table_index_create.rs

@@ -64,6 +64,7 @@ impl Interpreter for CreateTableIndexInterpreter {
             ast::TableIndexType::Inverted => TableIndexType::Inverted,
             ast::TableIndexType::Ngram => TableIndexType::Ngram,
             ast::TableIndexType::Vector => TableIndexType::Vector,
+            ast::TableIndexType::Spatial => TableIndexType::Spatial,
         };
 
         let create_index_req = CreateTableIndexReq {

src/query/service/src/interpreters/interpreter_table_index_drop.rs

@@ -61,6 +61,7 @@ impl Interpreter for DropTableIndexInterpreter {
             ast::TableIndexType::Inverted => TableIndexType::Inverted,
             ast::TableIndexType::Ngram => TableIndexType::Ngram,
             ast::TableIndexType::Vector => TableIndexType::Vector,
+            ast::TableIndexType::Spatial => TableIndexType::Spatial,
         };
 
         let drop_index_req = DropTableIndexReq {

src/query/service/src/interpreters/interpreter_table_index_refresh.rs

@@ -93,6 +93,7 @@ impl Interpreter for RefreshTableIndexInterpreter {
             ast::TableIndexType::Inverted => TableIndexType::Inverted,
             ast::TableIndexType::Ngram => TableIndexType::Ngram,
             ast::TableIndexType::Vector => TableIndexType::Vector,
+            ast::TableIndexType::Spatial => todo!(),
             ast::TableIndexType::Aggregating => unreachable!(),
         };
 

src/query/sql/src/planner/binder/ddl/index.rs

@@ -483,6 +483,12 @@ impl Binder {
                 let index_options = self.validate_vector_index_options(index_options)?;
                 (column_ids, index_options, TableIndexType::Vector)
             }
+            AstTableIndexType::Spatial => {
+                let column_ids =
+                    self.validate_spatial_index_columns(table_schema.clone(), columns)?;
+                let index_options = self.validate_spatial_index_options(index_options)?;
+                (column_ids, index_options, TableIndexType::Spatial)
+            }
             AstTableIndexType::Aggregating => unreachable!(),
         };
 
@@ -831,6 +837,52 @@ impl Binder {
         Ok(options)
     }
 
+    pub(in crate::planner::binder) fn validate_spatial_index_columns(
+        &self,
+        table_schema: TableSchemaRef,
+        columns: &[Identifier],
+    ) -> Result<Vec<ColumnId>> {
+        let mut column_set = BTreeSet::new();
+        for column in columns {
+            match table_schema.field_with_name(&column.name) {
+                Ok(field) => {
+                    if !matches!(
+                        field.data_type.remove_nullable(),
+                        TableDataType::Geometry | TableDataType::Geography
+                    ) {
+                        return Err(ErrorCode::UnsupportedIndex(format!(
+                            "Spatial index only support Geometry and Geography type, but the type of column {} is {}",
+                            column, field.data_type
+                        )));
+                    }
+                    if column_set.contains(&field.column_id) {
+                        return Err(ErrorCode::UnsupportedIndex(format!(
+                            "Spatial index column must be unique, but column {} is duplicate",
+                            column.name
+                        )));
+                    }
+                    column_set.insert(field.column_id);
+                }
+                Err(_) => {
+                    return Err(ErrorCode::UnsupportedIndex(format!(
+                        "Table does not have column {}",
+                        column
+                    )));
+                }
+            }
+        }
+        Ok(Vec::from_iter(column_set))
+    }
+
+    pub(in crate::planner::binder) fn validate_spatial_index_options(
+        &self,
+        _index_options: &BTreeMap<String, String>,
+    ) -> Result<BTreeMap<String, String>> {
+        let options = BTreeMap::new();
+        // todo
+        Ok(options)
+    }
+
     #[async_backtrace::framed]
     pub(in crate::planner::binder) async fn bind_drop_table_index(
         &mut self,

src/query/sql/src/planner/binder/ddl/table.rs

@@ -2081,6 +2081,15 @@ impl Binder {
                         self.validate_vector_index_options(&table_index_def.index_options)?;
                     (TableIndexType::Vector, column_ids, options)
                 }
+                AstTableIndexType::Spatial => {
+                    let column_ids = self.validate_spatial_index_columns(
+                        table_schema.clone(),
+                        &table_index_def.columns,
+                    )?;
+                    let options =
+                        self.validate_spatial_index_options(&table_index_def.index_options)?;
+                    (TableIndexType::Spatial, column_ids, options)
+                }
                 AstTableIndexType::Aggregating => unreachable!(),
             };
 

tests/sqllogictests/suites/query/index/10_spatial_index/10_0000_spatial_index_base.test

@@ -0,0 +1,40 @@
+statement ok
+drop database if exists test_spatial_index
+
+statement ok
+create database test_spatial_index
+
+statement ok
+use test_spatial_index
+
+statement ok
+CREATE TABLE IF NOT EXISTS t(id Int, geom Geometry, geog Geography, SPATIAL INDEX idx (geom, geog)) Engine = Fuse
+
+query TT
+SHOW CREATE TABLE t
+----
+t CREATE TABLE t ( id INT NULL, geom GEOMETRY NULL, geog GEOGRAPHY NULL, SYNC SPATIAL INDEX idx (geom, geog) ) ENGINE=FUSE
+
+statement ok
+DROP SPATIAL INDEX idx ON t;
+
+statement error
+CREATE SPATIAL INDEX idx2 ON t(id);
+
+statement ok
+CREATE SPATIAL INDEX idx2 ON t(geom, geog);
+
+query TT
+SHOW CREATE TABLE t
+----
+t CREATE TABLE t ( id INT NULL, geom GEOMETRY NULL, geog GEOGRAPHY NULL, SYNC SPATIAL INDEX idx2 (geom, geog) ) ENGINE=FUSE
+
+statement error
+DROP INVERTED INDEX idx2 ON t;
+
+statement ok
+use default
+
+statement ok
+drop database test_spatial_index
+

tests/suites/1_stateful/01_streaming_load/01_0001_streaming_load_ontime.result

@@ -2,29 +2,29 @@
 ontime_200.csv
 199
 
-199	2020	769
+199	2020.0	769
 --ontime_200.csv.gz
 ontime_200.csv.gz
 199
 
-199	2020	769
+199	2020.0	769
 --ontime_200.csv.zst
 ontime_200.csv.zst
 199
 
-199	2020	769
+199	2020.0	769
 --ontime_200.csv.bz2
 ontime_200.csv.bz2
 199
 
-199	2020	769
+199	2020.0	769
 --ontime_200.ndjson
 ontime_200.ndjson
 199
 
-199	2020	769
+199	2020.0	769
 --ontime_200.parquet
 ontime_200.parquet
 199
 
-199	2020	769
+199	2020.0	769

tests/suites/1_stateful/04_mini_dataset/04_0000_mini_ontime.result

@@ -34,7 +34,7 @@ MQ	12316.754450296687
 NW	11370.336373327018
 OH	12599.259602036094
 2006	12.16979169791698
-99999
+99999.0
 2006	1
 San Diego, CA	Los Angeles, CA	622
 Los Angeles, CA	San Diego, CA	619