fix(datafusion): normalize index_type in global index procedures

crates/integrations/datafusion/src/procedures.rs

@@ -28,6 +28,9 @@
 //! - `CALL sys.create_global_index(table => '...', index_column => '...', index_type => 'ivf-pq')`
 //! - `CALL sys.drop_global_index(table => '...', index_column => '...', index_type => 'btree')` (also 'bitmap', 'lumina', or a vindex type such as 'ivf-pq')
 //! - `CALL sys.create_lumina_index(table => '...', index_column => '...')`
+//!
+//! The `index_type` argument of the three global index procedures is
+//! case-insensitive and surrounding whitespace is ignored.
 
 use std::collections::HashMap;
 use std::sync::Arc;
@@ -42,6 +45,7 @@ use datafusion::sql::sqlparser::ast::{
     FunctionArguments, ObjectName, Value as SqlValue,
 };
 use paimon::catalog::{Catalog, Identifier};
+use paimon::lumina::LUMINA_IDENTIFIER;
 use paimon::spec::Snapshot;
 use paimon::table::{
     normalize_global_index_type_for_drop, SnapshotManager, Table, TagManager,
@@ -51,6 +55,10 @@ use paimon::vindex::is_vindex_index_type;
 
 use crate::error::to_datafusion_error;
 
+/// Default `index_type` for the global index procedures when the argument is
+/// omitted, matching Java's `CreateGlobalIndexProcedure`.
+const DEFAULT_GLOBAL_INDEX_TYPE: &str = "btree";
+
 /// Resolve a snapshot by id: try live snapshot file first, then fall back to tag metadata.
 async fn resolve_snapshot_by_id(
     sm: &SnapshotManager,
@@ -527,9 +535,12 @@ async fn proc_create_lumina_index(
     let index_column = require_arg(args, "index_column")?;
     let mut builder = table.new_lumina_index_build_builder();
     builder.with_index_column(index_column);
-    if let Some(index_type) = args.get("index_type") {
-        builder.with_index_type(index_type);
-    }
+    let index_type = normalize_index_type(
+        args.get("index_type")
+            .map(String::as_str)
+            .unwrap_or(LUMINA_IDENTIFIER),
+    );
+    builder.with_index_type(&index_type);
     if let Some(options) = args.get("options") {
         builder.with_options(parse_key_value_options(options)?);
     }
@@ -545,10 +556,12 @@ async fn proc_create_global_index(
 ) -> DFResult<DataFrame> {
     let table = get_table(catalog, catalog_name, args).await?;
     let index_column = require_arg(args, "index_column")?;
-    let index_type = args
+    let index_type_arg = args
         .get("index_type")
         .map(String::as_str)
-        .unwrap_or("btree");
+        .unwrap_or(DEFAULT_GLOBAL_INDEX_TYPE);
+    let index_type = normalize_index_type(index_type_arg);
+    let index_type = index_type.as_str();
     if is_sorted_global_index_type(index_type) {
         if args.contains_key("options") {
             return Err(DataFusionError::NotImplemented(
@@ -569,9 +582,10 @@ async fn proc_create_global_index(
         }
         builder.execute().await.map_err(to_datafusion_error)?;
     } else {
+        // Echo the raw argument, not the normalized one, so a typo stays visible.
         return Err(DataFusionError::NotImplemented(format!(
             "create_global_index only supports index_type => 'btree', 'bitmap', or vindex types \
-             ('ivf-flat', 'ivf-pq'), got '{index_type}'"
+             ('ivf-flat', 'ivf-pq'), got '{index_type_arg}'"
         )));
     }
     ok_result(ctx)
@@ -585,13 +599,16 @@ async fn proc_drop_global_index(
 ) -> DFResult<DataFrame> {
     let table = get_table(catalog, catalog_name, args).await?;
     let index_column = require_arg(args, "index_column")?;
-    let index_type = args
+    let index_type_arg = args
         .get("index_type")
         .map(String::as_str)
-        .unwrap_or("btree");
+        .unwrap_or(DEFAULT_GLOBAL_INDEX_TYPE);
+    let index_type = normalize_index_type(index_type_arg);
+    let index_type = index_type.as_str();
     if normalize_global_index_type_for_drop(index_type).is_none() {
+        // Echo the raw argument, not the normalized one, so a typo stays visible.
         return Err(DataFusionError::NotImplemented(format!(
-            "unsupported global index type '{index_type}'; supported: {SUPPORTED_GLOBAL_INDEX_TYPES_FOR_DROP}"
+            "unsupported global index type '{index_type_arg}'; supported: {SUPPORTED_GLOBAL_INDEX_TYPES_FOR_DROP}"
         )));
     }
     if args.contains_key("partitions") {
@@ -612,8 +629,19 @@ async fn proc_drop_global_index(
     ok_result(ctx)
 }
 
+/// Precondition: `index_type` is already canonical (see `normalize_index_type`).
 fn is_sorted_global_index_type(index_type: &str) -> bool {
-    index_type.eq_ignore_ascii_case("btree") || index_type.eq_ignore_ascii_case("bitmap")
+    index_type == "btree" || index_type == "bitmap"
+}
+
+/// Canonicalize a procedure's `index_type` argument: trim, then lowercase.
+/// Mirrors `indexType.toLowerCase(Locale.ROOT).trim()` in Java's Flink and Spark
+/// `CreateGlobalIndexProcedure` / `DropGlobalIndexProcedure`. Normalizing at this
+/// boundary keeps the core builders' exact matching intact -- they are the analog
+/// of Java's `GlobalIndexer`, which likewise receives an already-canonical value
+/// and persists it into index metadata.
+fn normalize_index_type(index_type: &str) -> String {
+    index_type.trim().to_ascii_lowercase()
 }
 
 fn parse_key_value_options(options: &str) -> DFResult<HashMap<String, String>> {
@@ -768,4 +796,33 @@ mod tests {
         let result = earlier_or_equal_from_all(&sm, &tm, 1500).await.unwrap();
         assert_eq!(result.unwrap().id(), 1);
     }
+
+    #[test]
+    fn test_normalize_index_type() {
+        // Casing and surrounding whitespace are both absorbed, matching Java's
+        // `indexType.toLowerCase(Locale.ROOT).trim()`.
+        assert_eq!(normalize_index_type("BTREE"), "btree");
+        assert_eq!(normalize_index_type(" btree "), "btree");
+        assert_eq!(normalize_index_type(" BitMap\t"), "bitmap");
+        assert_eq!(normalize_index_type("IVF-FLAT"), "ivf-flat");
+        assert_eq!(normalize_index_type("Ivf-Pq"), "ivf-pq");
+        assert_eq!(
+            normalize_index_type("Lumina-Vector-Ann"),
+            "lumina-vector-ann"
+        );
+        // Already canonical values are returned unchanged, and an unknown type
+        // is normalized but not rewritten -- the caller still rejects it.
+        assert_eq!(normalize_index_type("ivf-flat"), "ivf-flat");
+        assert_eq!(normalize_index_type(" Full-Text "), "full-text");
+    }
+
+    #[test]
+    fn test_sorted_global_index_type_predicate() {
+        assert!(is_sorted_global_index_type("btree"));
+        assert!(is_sorted_global_index_type("bitmap"));
+        assert!(!is_sorted_global_index_type("ivf-flat"));
+        assert!(!is_sorted_global_index_type("lumina"));
+        // The predicate requires a canonical input; callers normalize first.
+        assert!(!is_sorted_global_index_type("BTREE"));
+    }
 }

crates/integrations/datafusion/tests/procedures.rs

@@ -140,6 +140,38 @@ async fn setup_btree_global_index_table(
     (tmp, sql_context)
 }
 
+/// An `ARRAY<FLOAT>` table configured so the pure-Rust `ivf-flat` builder can
+/// train and commit an index without a native library.
+async fn setup_vindex_global_index_table(
+    table_name: &str,
+) -> (tempfile::TempDir, paimon_datafusion::SQLContext) {
+    let (tmp, sql_context) = setup_sql_context().await;
+    exec(
+        &sql_context,
+        &format!(
+            "CREATE TABLE paimon.test_db.{table_name} (id INT, embedding ARRAY<FLOAT>) WITH (\
+                'row-tracking.enabled' = 'true',\
+                'data-evolution.enabled' = 'true',\
+                'global-index.enabled' = 'true',\
+                'global-index.row-count-per-shard' = '100',\
+                'ivf-flat.dimension' = '2',\
+                'ivf-flat.nlist' = '1',\
+                'ivf-flat.distance.metric' = 'l2'\
+            )"
+        ),
+    )
+    .await;
+    exec(
+        &sql_context,
+        &format!(
+            "INSERT INTO paimon.test_db.{table_name} (id, embedding) VALUES \
+             (1, [1.0, 0.0]), (2, [0.0, 1.0]), (3, [1.0, 1.0])"
+        ),
+    )
+    .await;
+    (tmp, sql_context)
+}
+
 #[tokio::test]
 async fn test_create_global_index_requires_index_column() {
     let (_tmp, sql_context) = setup_btree_global_index_table("btree_missing_col").await;
@@ -184,6 +216,89 @@ async fn test_create_global_index_rejects_options() {
     .await;
 }
 
+/// `index_type` was matched case-insensitively for btree/bitmap but exactly for
+/// the vindex types, so `'IVF-FLAT'` was rejected as unsupported while `'BTREE'`
+/// worked. Build with an uppercase vindex type and assert the index lands with
+/// the canonical lowercase `index_type` in the manifest, which is what the read
+/// path matches on.
+#[tokio::test]
+async fn test_create_global_index_accepts_uppercase_vindex_type() {
+    let (_tmp, sql_context) = setup_vindex_global_index_table("vindex_upper").await;
+
+    exec(
+        &sql_context,
+        "CALL sys.create_global_index(table => 'test_db.vindex_upper', index_column => 'embedding', index_type => 'IVF-FLAT')",
+    )
+    .await;
+
+    let index_count = row_count(
+        &sql_context,
+        "SELECT * FROM paimon.test_db.`vindex_upper$table_indexes` \
+         WHERE index_type = 'ivf-flat' AND index_field_name = 'embedding'",
+    )
+    .await;
+    assert_eq!(index_count, 1);
+}
+
+/// Neither procedure trimmed the argument, so `' btree '` was reported as an
+/// unsupported type. Both sides now absorb surrounding whitespace, matching
+/// Java's `toLowerCase().trim()`.
+#[tokio::test]
+async fn test_global_index_procedures_ignore_surrounding_whitespace() {
+    let (_tmp, sql_context) = setup_btree_global_index_table("btree_trim").await;
+    exec(
+        &sql_context,
+        "INSERT INTO paimon.test_db.btree_trim (id, name) VALUES (1, 'alice'), (2, 'bob')",
+    )
+    .await;
+
+    exec(
+        &sql_context,
+        "CALL sys.create_global_index(table => 'test_db.btree_trim', index_column => 'id', index_type => ' btree ')",
+    )
+    .await;
+    let after_create = row_count(
+        &sql_context,
+        "SELECT * FROM paimon.test_db.`btree_trim$table_indexes` \
+         WHERE index_type = 'btree' AND index_field_name = 'id'",
+    )
+    .await;
+    assert_eq!(after_create, 1);
+
+    exec(
+        &sql_context,
+        "CALL sys.drop_global_index(table => 'test_db.btree_trim', index_column => 'id', index_type => ' BTREE ')",
+    )
+    .await;
+    let after_drop = row_count(
+        &sql_context,
+        "SELECT * FROM paimon.test_db.`btree_trim$table_indexes` \
+         WHERE index_type = 'btree' AND index_field_name = 'id'",
+    )
+    .await;
+    assert_eq!(after_drop, 0);
+}
+
+/// An unsupported type must be echoed back exactly as written, not lowercased,
+/// so a typo stays recognizable in the error message.
+#[tokio::test]
+async fn test_global_index_procedures_echo_raw_unsupported_type() {
+    let (_tmp, sql_context) = setup_btree_global_index_table("btree_echo").await;
+
+    assert_sql_error(
+        &sql_context,
+        "CALL sys.create_global_index(table => 'test_db.btree_echo', index_column => 'id', index_type => 'Full-Text')",
+        "got 'Full-Text'",
+    )
+    .await;
+    assert_sql_error(
+        &sql_context,
+        "CALL sys.drop_global_index(table => 'test_db.btree_echo', index_column => 'id', index_type => 'Full-Text')",
+        "unsupported global index type 'Full-Text'",
+    )
+    .await;
+}
+
 #[tokio::test]
 async fn test_create_global_index_builds_btree_and_filter_reads() {
     let (_tmp, sql_context) = setup_btree_global_index_table("btree_build").await;

docs/src/sql.md

@@ -964,10 +964,10 @@ CALL sys.create_global_index(
 );
 ```
 
-`index_type` defaults to `btree`. BTree and bitmap global indexes support
-scalar columns and do not accept the `options` argument yet. Bitmap global
-indexes use the same on-disk file format as Java Paimon's
-`BitmapGlobalIndexFormat`.
+`index_type` defaults to `btree`. It is case-insensitive and surrounding
+whitespace is ignored. BTree and bitmap global indexes support scalar columns
+and do not accept the `options` argument yet. Bitmap global indexes use the same
+on-disk file format as Java Paimon's `BitmapGlobalIndexFormat`.
 
 The current global-index builders require a row-tracking data-evolution table
 with global indexes enabled. They do not support primary-key tables or tables
@@ -1052,7 +1052,8 @@ CALL sys.drop_global_index(
 
 `index_type` accepts every type the create procedures build: `btree`, `bitmap`,
 `lumina` (or `lumina-vector-ann`), and the vindex types `ivf-flat` and `ivf-pq`.
-It defaults to `btree`.
+It defaults to `btree`, is case-insensitive and surrounding whitespace is
+ignored.
 
 ### create_lumina_index
 
@@ -1062,8 +1063,9 @@ Build and commit a Lumina global vector index for a table column:
 CALL sys.create_lumina_index(table => 'paimon.my_db.my_table', index_column => 'embedding');
 ```
 
-The optional `index_type` argument selects the Lumina index identifier. It defaults to
-`lumina`. Valid values are `lumina` and the legacy-compatible `lumina-vector-ann`.
+The optional `index_type` argument selects the Lumina index identifier. It
+defaults to `lumina`, is case-insensitive and surrounding whitespace is ignored.
+Valid values are `lumina` and the legacy-compatible `lumina-vector-ann`.
 
 ```sql
 CALL sys.create_lumina_index(