fix(scan): respect family-specific global-index search modes

crates/paimon/src/spec/core_options.rs

@@ -23,6 +23,9 @@ pub(crate) const QUERY_AUTH_ENABLED_OPTION: &str = "query-auth.enabled";
 const DATA_EVOLUTION_ENABLED_OPTION: &str = "data-evolution.enabled";
 const GLOBAL_INDEX_ENABLED_OPTION: &str = "global-index.enabled";
 const GLOBAL_INDEX_SEARCH_MODE_OPTION: &str = "global-index.search-mode";
+const SCALAR_INDEX_SEARCH_MODE_OPTION: &str = "scalar-index.search-mode";
+const VECTOR_INDEX_SEARCH_MODE_OPTION: &str = "vector-index.search-mode";
+const FULL_TEXT_INDEX_SEARCH_MODE_OPTION: &str = "full-text-index.search-mode";
 const GLOBAL_INDEX_ROW_COUNT_PER_SHARD_OPTION: &str = "global-index.row-count-per-shard";
 const GLOBAL_INDEX_THREAD_NUM_OPTION: &str = "global-index.thread-num";
 const GLOBAL_INDEX_COLUMN_UPDATE_ACTION_OPTION: &str = "global-index.column-update-action";
@@ -639,18 +642,36 @@ impl<'a> CoreOptions<'a> {
     }
 
     pub fn global_index_search_mode(&self) -> crate::Result<GlobalIndexSearchMode> {
-        match self
-            .options
-            .get(GLOBAL_INDEX_SEARCH_MODE_OPTION)
-            .map(|v| v.to_ascii_lowercase())
-            .as_deref()
-            .unwrap_or("fast")
-        {
+        self.index_search_mode(GLOBAL_INDEX_SEARCH_MODE_OPTION)
+    }
+
+    pub fn scalar_index_search_mode(&self) -> crate::Result<GlobalIndexSearchMode> {
+        self.index_search_mode(SCALAR_INDEX_SEARCH_MODE_OPTION)
+    }
+
+    pub fn vector_index_search_mode(&self) -> crate::Result<GlobalIndexSearchMode> {
+        self.index_search_mode(VECTOR_INDEX_SEARCH_MODE_OPTION)
+    }
+
+    pub fn full_text_index_search_mode(&self) -> crate::Result<GlobalIndexSearchMode> {
+        self.index_search_mode(FULL_TEXT_INDEX_SEARCH_MODE_OPTION)
+    }
+
+    fn index_search_mode(&self, family_option: &str) -> crate::Result<GlobalIndexSearchMode> {
+        let (option, value) = if let Some(value) = self.options.get(family_option) {
+            (family_option, value)
+        } else if let Some(value) = self.options.get(GLOBAL_INDEX_SEARCH_MODE_OPTION) {
+            (GLOBAL_INDEX_SEARCH_MODE_OPTION, value)
+        } else {
+            return Ok(GlobalIndexSearchMode::Fast);
+        };
+
+        match value.to_ascii_lowercase().as_str() {
             "fast" => Ok(GlobalIndexSearchMode::Fast),
             "full" => Ok(GlobalIndexSearchMode::Full),
             "detail" => Ok(GlobalIndexSearchMode::Detail),
             other => Err(crate::Error::ConfigInvalid {
-                message: format!("Unsupported global-index.search-mode: {other}"),
+                message: format!("Unsupported {option}: {other}"),
             }),
         }
     }
@@ -1582,6 +1603,54 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_family_index_search_mode_precedence() {
+        let legacy = HashMap::from([(
+            GLOBAL_INDEX_SEARCH_MODE_OPTION.to_string(),
+            "full".to_string(),
+        )]);
+        let legacy_core = CoreOptions::new(&legacy);
+        for actual in [
+            legacy_core.scalar_index_search_mode().unwrap(),
+            legacy_core.vector_index_search_mode().unwrap(),
+            legacy_core.full_text_index_search_mode().unwrap(),
+        ] {
+            assert_eq!(actual, GlobalIndexSearchMode::Full);
+        }
+
+        let family = HashMap::from([
+            (
+                GLOBAL_INDEX_SEARCH_MODE_OPTION.to_string(),
+                "full".to_string(),
+            ),
+            (
+                SCALAR_INDEX_SEARCH_MODE_OPTION.to_string(),
+                "fast".to_string(),
+            ),
+            (
+                VECTOR_INDEX_SEARCH_MODE_OPTION.to_string(),
+                "detail".to_string(),
+            ),
+            (
+                FULL_TEXT_INDEX_SEARCH_MODE_OPTION.to_string(),
+                "fast".to_string(),
+            ),
+        ]);
+        let family_core = CoreOptions::new(&family);
+        assert_eq!(
+            family_core.scalar_index_search_mode().unwrap(),
+            GlobalIndexSearchMode::Fast
+        );
+        assert_eq!(
+            family_core.vector_index_search_mode().unwrap(),
+            GlobalIndexSearchMode::Detail
+        );
+        assert_eq!(
+            family_core.full_text_index_search_mode().unwrap(),
+            GlobalIndexSearchMode::Fast
+        );
+    }
+
     #[test]
     fn test_global_index_enabled_defaults_and_overrides() {
         assert!(CoreOptions::new(&HashMap::new()).global_index_enabled());

crates/paimon/src/table/btree_global_index_build_builder.rs

@@ -1432,6 +1432,72 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn test_scalar_full_search_includes_unindexed_rows() {
+        let mut options = table_options("2");
+        options.insert("scalar-index.search-mode".to_string(), "full".to_string());
+        let table = test_table_with_path(
+            "memory:/test_scalar_full_search_includes_unindexed_rows",
+            options,
+        );
+        setup_dirs(&table).await;
+
+        let mut first_write = TableWrite::new(&table, "writer-1".to_string()).unwrap();
+        first_write
+            .write_arrow_batch(&data_batch(vec![1, 2], vec!["alice", "bob"]))
+            .await
+            .unwrap();
+        TableCommit::new(table.clone(), "writer-1".to_string())
+            .commit(first_write.prepare_commit().await.unwrap())
+            .await
+            .unwrap();
+        table
+            .new_btree_global_index_build_builder()
+            .with_index_column("name")
+            .execute()
+            .await
+            .unwrap();
+
+        let mut second_write = TableWrite::new(&table, "writer-2".to_string()).unwrap();
+        second_write
+            .write_arrow_batch(&data_batch(vec![3, 4], vec!["alice", "dave"]))
+            .await
+            .unwrap();
+        TableCommit::new(table.clone(), "writer-2".to_string())
+            .commit(second_write.prepare_commit().await.unwrap())
+            .await
+            .unwrap();
+
+        let predicate = PredicateBuilder::new(table.schema().fields())
+            .equal("name", crate::spec::Datum::String("alice".to_string()))
+            .unwrap();
+        let mut read_builder = table.new_read_builder();
+        read_builder.with_filter(predicate);
+        let plan = read_builder.new_scan().plan().await.unwrap();
+        let planned_ranges = merge_row_ranges(
+            plan.splits()
+                .iter()
+                .flat_map(|split| split.row_ranges().unwrap_or_default())
+                .cloned()
+                .collect(),
+        );
+
+        assert_eq!(
+            planned_ranges,
+            vec![RowRange::new(0, 0), RowRange::new(2, 3)]
+        );
+        assert_eq!(
+            scan_ids(
+                &table,
+                PredicateBuilder::new(table.schema().fields())
+                    .equal("name", crate::spec::Datum::String("alice".to_string()))
+                    .unwrap(),
+            )
+            .await,
+            vec![1, 3]
+        );
+    }
+
     #[tokio::test]
     async fn test_empty_global_index_ranges_skip_legacy_manifests() {
         for search_mode in ["fast", "full"] {

crates/paimon/src/table/full_text_search_builder.rs

@@ -231,7 +231,7 @@ impl<'a> FullTextSearchBuilder<'a> {
         }
 
         // FAST-only: reject FULL/DETAIL loud rather than silently degrading.
-        if core.global_index_search_mode()? != GlobalIndexSearchMode::Fast {
+        if core.full_text_index_search_mode()? != GlobalIndexSearchMode::Fast {
             return Err(crate::Error::DataInvalid {
                 message: "primary-key full-text search supports only the FAST global-index search \
                           mode"
@@ -313,7 +313,7 @@ async fn evaluate_full_text_search(
 ) -> crate::Result<SearchResult> {
     let table_path = evaluation.table_path.trim_end_matches('/');
     let core_options = CoreOptions::new(evaluation.table_options);
-    let search_mode = core_options.global_index_search_mode()?;
+    let search_mode = core_options.full_text_index_search_mode()?;
 
     let field_id = match find_field_id_by_name(evaluation.schema_fields, &search.field_name) {
         Some(id) => id,
@@ -868,7 +868,10 @@ mod tests {
             DataType::Int(IntType::default()),
         )];
         let search = FullTextSearch::new("hello".to_string(), 10, "body".to_string()).unwrap();
-        let options = HashMap::from([("global-index.search-mode".to_string(), "full".to_string())]);
+        let options = HashMap::from([(
+            "full-text-index.search-mode".to_string(),
+            "full".to_string(),
+        )]);
 
         let result = evaluate_full_text_search(
             FullTextSearchEvaluation {

crates/paimon/src/table/hybrid_search_builder.rs

@@ -611,7 +611,7 @@ impl<'a> HybridSearchBuilder<'a> {
         // payloads and rejects FULL/DETAIL loud rather than silently degrading,
         // mirroring `full_text_search_builder::execute_read` and Java
         // `PrimaryKeyFullTextRead.checkFastSearchMode`.
-        if core.global_index_search_mode()? != GlobalIndexSearchMode::Fast {
+        if core.full_text_index_search_mode()? != GlobalIndexSearchMode::Fast {
             return Err(crate::Error::DataInvalid {
                 message: "primary-key full-text search supports only the FAST global-index search \
                           mode"
@@ -1785,7 +1785,7 @@ mod pk_hybrid_tests {
                 [7.0, 0.0, 0.0, 0.0],
             ],
             &["alpha", "beta", "alpha alpha alpha", "gamma"],
-            &[("global-index.search-mode", "full")],
+            &[("full-text-index.search-mode", "full")],
         )
         .await;
 

crates/paimon/src/table/table_scan.rs

@@ -1209,7 +1209,7 @@ impl<'a> PaimonTableScan<'a> {
             !self.data_predicates.is_empty(),
         ) {
             Ok(Some(GlobalIndexScanSettings {
-                search_mode: core_options.global_index_search_mode()?,
+                search_mode: core_options.scalar_index_search_mode()?,
                 thread_num: core_options.global_index_thread_num()?,
             }))
         } else {

crates/paimon/src/table/vector_search_builder.rs

@@ -762,7 +762,7 @@ async fn plan_and_search_pk_candidates_batch(
             source: None,
         })?;
 
-    let search_mode = core.global_index_search_mode()?;
+    let search_mode = core.vector_index_search_mode()?;
     let skip_exact_fallback = search_mode == GlobalIndexSearchMode::Fast;
 
     // A non-positive limit is invalid regardless of the plan; reject it before
@@ -1435,7 +1435,7 @@ async fn evaluate_batch_vector_search(
 
     let table_path = evaluation.table_path.trim_end_matches('/');
     let core_options = CoreOptions::new(evaluation.table_options);
-    let search_mode = core_options.global_index_search_mode()?;
+    let search_mode = core_options.vector_index_search_mode()?;
     let field_name = &vector_searches[0].field_name;
     if vector_searches
         .iter()
@@ -3487,7 +3487,7 @@ mod tests {
         let file_io = crate::io::FileIOBuilder::new("memory").build().unwrap();
         let fields = vec![make_field(2, "embedding")];
         let vs = VectorSearch::new(vec![1.0], 10, "embedding".to_string()).unwrap();
-        let options = HashMap::from([("global-index.search-mode".to_string(), "full".to_string())]);
+        let options = HashMap::from([("vector-index.search-mode".to_string(), "full".to_string())]);
 
         let err = evaluate_vector_search(
             eval_context(&file_io, &options, &fields, Some(10)),