feat: use StrinView to speed up query for main (#7400)

Tow new env
```
ZO_SEARCH_INSPECTOR_ENABLED = false
ZO_UTF8_VIEW_ENABLED = true
```

src/config/src/config.rs

@@ -1143,6 +1143,10 @@ pub struct Common {
     pub use_stream_settings_for_partitions_enabled: bool,
     #[env_config(name = "ZO_DASHBOARD_PLACEHOLDER", default = "_o2_all_")]
     pub dashboard_placeholder: String,
+    #[env_config(name = "ZO_SEARCH_INSPECTOR_ENABLED", default = false)]
+    pub search_inspector_enabled: bool,
+    #[env_config(name = "ZO_UTF8_VIEW_ENABLED", default = true)]
+    pub utf8_view_enabled: bool,
 }
 
 #[derive(EnvConfig)]

src/service/search/datafusion/optimizer/rewrite_match.rs

@@ -13,6 +13,7 @@
 // You should have received a copy of the GNU Affero General Public License
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+use config::get_config;
 use datafusion::{
     self,
     common::{
@@ -127,7 +128,11 @@ impl TreeNodeRewriter for MatchToFullTextMatch {
                         .trim_start_matches('*') // contains
                         .trim_end_matches('*') // prefix or contains
                         .to_string(); // remove prefix and suffix *
-                    let item = Expr::Literal(ScalarValue::Utf8(Some(format!("%{item}%"))));
+                    let item = if get_config().common.utf8_view_enabled {
+                        Expr::Literal(ScalarValue::Utf8View(Some(format!("%{item}%"))))
+                    } else {
+                        Expr::Literal(ScalarValue::Utf8(Some(format!("%{item}%"))))
+                    };
                     for field in self.fields.iter() {
                         let new_expr = Expr::Like(Like {
                             negated: false,
@@ -159,7 +164,11 @@ impl TreeNodeRewriter for MatchToFullTextMatch {
                         )));
                     };
                     let mut expr_list = Vec::with_capacity(self.fields.len());
-                    let item = Expr::Literal(ScalarValue::Utf8(Some(item.to_string())));
+                    let item = if get_config().common.utf8_view_enabled {
+                        Expr::Literal(ScalarValue::Utf8View(Some(item.to_string())))
+                    } else {
+                        Expr::Literal(ScalarValue::Utf8(Some(item.to_string())))
+                    };
                     let distance = Expr::Literal(ScalarValue::Int64(Some(distance)));
                     let fuzzy_expr = fuzzy_match_udf::FUZZY_MATCH_UDF.clone();
                     for field in self.fields.iter() {
@@ -190,8 +199,9 @@ impl TreeNodeRewriter for MatchToFullTextMatch {
 mod tests {
     use std::sync::Arc;
 
-    use arrow::array::{Int64Array, StringArray};
+    use arrow::array::{Array, Int64Array, StringArray, StringViewArray};
     use arrow_schema::DataType;
+    use config::get_config;
     use datafusion::{
         arrow::{
             datatypes::{Field, Schema},
@@ -234,32 +244,64 @@ mod tests {
             ),
         ];
 
-        // define a schema.
-        let schema = Arc::new(Schema::new(vec![
-            Field::new("_timestamp", DataType::Int64, false),
-            Field::new("name", DataType::Utf8, false),
-            Field::new("log", DataType::Utf8, false),
-        ]));
+        let schema = if get_config().common.utf8_view_enabled {
+            // define a schema.
+            Arc::new(Schema::new(vec![
+                Field::new("_timestamp", DataType::Int64, false),
+                Field::new("name", DataType::Utf8View, false),
+                Field::new("log", DataType::Utf8View, false),
+            ]))
+        } else {
+            Arc::new(Schema::new(vec![
+                Field::new("_timestamp", DataType::Int64, false),
+                Field::new("name", DataType::Utf8, false),
+                Field::new("log", DataType::Utf8, false),
+            ]))
+        };
+
+        let name_array: Arc<dyn Array> = if get_config().common.utf8_view_enabled {
+            Arc::new(StringViewArray::from(vec![
+                "open",
+                "observe",
+                "openobserve",
+                "OBserve",
+                "oo",
+            ]))
+        } else {
+            Arc::new(StringArray::from(vec![
+                "open",
+                "observe",
+                "openobserve",
+                "OBserve",
+                "oo",
+            ]))
+        };
+
+        let log_array: Arc<dyn Array> = if get_config().common.utf8_view_enabled {
+            Arc::new(StringViewArray::from(vec![
+                "o2",
+                "obSERVE",
+                "openobserve",
+                "o2",
+                "oo",
+            ]))
+        } else {
+            Arc::new(StringArray::from(vec![
+                "o2",
+                "obSERVE",
+                "openobserve",
+                "o2",
+                "oo",
+            ]))
+        };
 
         // define data.
         let batch = RecordBatch::try_new(
             schema.clone(),
             vec![
                 Arc::new(Int64Array::from(vec![1, 2, 3, 4, 5])),
-                Arc::new(StringArray::from(vec![
-                    "open",
-                    "observe",
-                    "openobserve",
-                    "OBserve",
-                    "oo",
-                ])),
-                Arc::new(StringArray::from(vec![
-                    "o2",
-                    "obSERVE",
-                    "openobserve",
-                    "o2",
-                    "oo",
-                ])),
+                name_array,
+                log_array,
             ],
         )
         .unwrap();

src/service/search/datafusion/udf/mod.rs

@@ -55,15 +55,7 @@ pub(crate) const REGEX_NOT_MATCH_UDF_NAME: &str = "re_not_match";
 /// The name of the regex_matches UDF given to DataFusion.
 pub(crate) const REGEX_MATCHES_UDF_NAME: &str = "re_matches";
 
-pub(crate) const DEFAULT_FUNCTIONS: [ZoFunction; 13] = [
-    ZoFunction {
-        name: "match_all_raw",
-        text: "match_all_raw('v')",
-    },
-    ZoFunction {
-        name: "match_all_raw_ignore_case",
-        text: "match_all_raw_ignore_case('v')",
-    },
+pub(crate) const DEFAULT_FUNCTIONS: [ZoFunction; 11] = [
     ZoFunction {
         name: "match_all",
         text: "match_all('v')",

src/service/search/grpc/mod.rs

@@ -21,6 +21,7 @@ use infra::errors::Result;
 
 pub mod flight;
 pub mod storage;
+pub(crate) mod utils;
 pub mod wal;
 
 pub type SearchTable = Result<(Vec<Arc<dyn TableProvider>>, ScanStats)>;

src/service/search/grpc/storage.rs

@@ -53,6 +53,7 @@ use crate::service::{
     search::{
         datafusion::exec,
         generate_search_schema_diff,
+        grpc::utils,
         index::IndexCondition,
         inspector::{SearchInspectorFieldsBuilder, search_inspector_fields},
         tantivy::puffin_directory::{
@@ -350,8 +351,10 @@ pub async fn search(
         if files.is_empty() {
             continue;
         }
-        let schema = schema_versions[ver].clone();
-        let schema = schema.with_metadata(Default::default());
+        let schema = schema_versions[ver]
+            .clone()
+            .with_metadata(Default::default());
+        let schema = utils::change_schema_to_utf8_view(schema);
 
         let session = config::meta::search::Session {
             id: format!("{}-storage-{ver}", query.trace_id),

src/service/search/grpc/utils.rs

@@ -0,0 +1,38 @@
+// Copyright 2025 OpenObserve Inc.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+use std::sync::Arc;
+
+use arrow_schema::{DataType, Field, Schema};
+use config::get_config;
+
+pub fn change_schema_to_utf8_view(schema: Schema) -> Schema {
+    if !get_config().common.utf8_view_enabled {
+        return schema;
+    }
+
+    let fields = schema
+        .fields()
+        .iter()
+        .map(|f| {
+            if f.data_type() == &DataType::Utf8 {
+                Arc::new(Field::new(f.name(), DataType::Utf8View, f.is_nullable()))
+            } else {
+                f.clone()
+            }
+        })
+        .collect::<Vec<_>>();
+    Schema::new(fields)
+}

src/service/search/grpc/wal.rs

@@ -16,6 +16,7 @@
 use std::{path::Path, sync::Arc};
 
 use arrow::array::{ArrayRef, new_null_array};
+use arrow_schema::{DataType, Field};
 use config::{
     cluster::LOCAL_NODE,
     get_config,
@@ -46,6 +47,7 @@ use crate::{
         search::{
             datafusion::{exec, table_provider::memtable::NewMemTable},
             generate_filter_from_equal_items, generate_search_schema_diff,
+            grpc::utils,
             index::IndexCondition,
             inspector::{SearchInspectorFieldsBuilder, search_inspector_fields},
             match_source,
@@ -275,6 +277,8 @@ pub async fn search_parquet(
         let schema = schema_versions[ver]
             .clone()
             .with_metadata(Default::default());
+        let schema = utils::change_schema_to_utf8_view(schema);
+
         let session = config::meta::search::Session {
             id: format!("{}-wal-{ver}", query.trace_id),
             storage_type: StorageType::Wal,
@@ -438,10 +442,13 @@ pub async fn search_memtable(
             continue;
         }
 
-        let diff_fields = generate_search_schema_diff(&schema, &latest_schema_map);
+        // the field in latest_schema_map, but not in schema,
+        // so not in the diff_fields, result in Utf8View issue
+        // so we need to add it to the diff_fields
+        let mut diff_fields = generate_search_schema_diff(&schema, &latest_schema_map);
         let mut adapt_batches = Vec::with_capacity(record_batches.len());
         for batch in record_batches {
-            adapt_batches.push(adapt_batch(latest_schema.clone(), batch));
+            adapt_batches.push(adapt_batch(latest_schema.clone(), batch, &mut diff_fields));
         }
         let record_batches = adapt_batches;
 
@@ -644,7 +651,11 @@ async fn get_file_list(
     .await
 }
 
-pub fn adapt_batch(latest_schema: Arc<Schema>, batch: RecordBatch) -> RecordBatch {
+fn adapt_batch(
+    latest_schema: Arc<Schema>,
+    batch: RecordBatch,
+    diff_fields: &mut HashMap<String, DataType>,
+) -> RecordBatch {
     let batch_schema = &*batch.schema();
     let batch_cols = batch.columns().to_vec();
 
@@ -654,6 +665,15 @@ pub fn adapt_batch(latest_schema: Arc<Schema>, batch: RecordBatch) -> RecordBatc
         if let Some((idx, field)) = batch_schema.column_with_name(field_latest.name()) {
             cols.push(Arc::clone(&batch_cols[idx]));
             fields.push(field.clone());
+        } else if *field_latest.data_type() == DataType::Utf8View {
+            // in memtable, the schema should be utf8
+            cols.push(new_null_array(&DataType::Utf8, batch.num_rows()));
+            fields.push(Field::new(
+                field_latest.name(),
+                DataType::Utf8,
+                field_latest.is_nullable(),
+            ));
+            diff_fields.insert(field_latest.name().to_string(), DataType::Utf8View);
         } else {
             cols.push(new_null_array(field_latest.data_type(), batch.num_rows()));
             fields.push(field_latest.as_ref().clone());

src/service/search/index.rs

@@ -19,7 +19,7 @@ use std::{
 };
 
 use config::{
-    INDEX_FIELD_NAME_FOR_ALL,
+    INDEX_FIELD_NAME_FOR_ALL, get_config,
     utils::tantivy::{query::contains_query::ContainsQuery, tokenizer::o2_collect_tokens},
 };
 use datafusion::{
@@ -563,7 +563,13 @@ impl Condition {
                     .trim_start_matches('*') // contains
                     .trim_end_matches('*') // prefix or contains
                     .to_string();
-                let term = Arc::new(Literal::new(ScalarValue::Utf8(Some(format!("%{value}%")))));
+                let term = if get_config().common.utf8_view_enabled {
+                    Arc::new(Literal::new(ScalarValue::Utf8View(Some(format!(
+                        "%{value}%"
+                    )))))
+                } else {
+                    Arc::new(Literal::new(ScalarValue::Utf8(Some(format!("%{value}%")))))
+                };
                 let mut expr_list: Vec<Arc<dyn PhysicalExpr>> =
                     Vec::with_capacity(fst_fields.len());
                 for field in fst_fields.iter() {
@@ -584,7 +590,11 @@ impl Condition {
             }
             Condition::FuzzyMatchAll(value, distance) => {
                 let fuzzy_expr = Arc::new(fuzzy_match_udf::FUZZY_MATCH_UDF.clone());
-                let term = Arc::new(Literal::new(ScalarValue::Utf8(Some(value.clone()))));
+                let term = if get_config().common.utf8_view_enabled {
+                    Arc::new(Literal::new(ScalarValue::Utf8View(Some(value.to_string()))))
+                } else {
+                    Arc::new(Literal::new(ScalarValue::Utf8(Some(value.to_string()))))
+                };
                 let distance = Arc::new(Literal::new(ScalarValue::Int64(Some(*distance as i64))));
                 let mut expr_list: Vec<Arc<dyn PhysicalExpr>> =
                     Vec::with_capacity(fst_fields.len());
@@ -779,6 +789,9 @@ fn get_scalar_value(value: &str, data_type: &DataType) -> Result<Arc<Literal>, a
         DataType::Binary => Arc::new(Literal::new(ScalarValue::Binary(Some(
             value.as_bytes().to_vec(),
         )))),
+        DataType::Utf8View => {
+            Arc::new(Literal::new(ScalarValue::Utf8View(Some(value.to_string()))))
+        }
         _ => unimplemented!(),
     })
 }

src/service/search/inspector.rs

@@ -13,7 +13,7 @@
 // You should have received a copy of the GNU Affero General Public License
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-use config::{cluster::LOCAL_NODE, meta::cluster::NodeInfo};
+use config::{cluster::LOCAL_NODE, get_config, meta::cluster::NodeInfo};
 use serde::{Deserialize, Serialize};
 
 #[derive(Debug, Default, Serialize, Deserialize)]
@@ -118,6 +118,14 @@ impl SearchInspectorFieldsBuilder {
 }
 
 pub fn search_inspector_fields(msg: String, kvs: SearchInspectorFields) -> String {
+    if !get_config().common.search_inspector_enabled {
+        return msg;
+    }
+
+    search_inspector_fields_inner(msg, kvs)
+}
+
+fn search_inspector_fields_inner(msg: String, kvs: SearchInspectorFields) -> String {
     if msg.is_empty() {
         return msg;
     }
@@ -154,7 +162,7 @@ mod tests {
             .component("search".to_string())
             .build();
 
-        let result = search_inspector_fields(msg.clone(), fields);
+        let result = search_inspector_fields_inner(msg.clone(), fields);
         assert!(result.contains("test message"));
         assert!(result.contains("\"duration\":100"));
         assert!(result.contains("\"component\":\"search\""));

src/service/search/sql.rs

@@ -15,7 +15,7 @@
 
 use std::{ops::ControlFlow, sync::Arc};
 
-use arrow_schema::FieldRef;
+use arrow_schema::{DataType, Field, FieldRef};
 use chrono::Duration;
 use config::{
     ALL_VALUES_COL_NAME, ID_COL_NAME, ORIGINAL_DATA_COL_NAME, TIMESTAMP_COL_NAME, get_config,
@@ -327,7 +327,33 @@ impl Sql {
             }
         }
 
+        // 15. replace the Utf8 to Utf8View type
+        let final_schemas = if cfg.common.utf8_view_enabled {
+            let mut final_schemas = HashMap::with_capacity(used_schemas.len());
+            for (stream, schema) in used_schemas.iter() {
+                let fields = schema
+                    .schema()
+                    .fields()
+                    .iter()
+                    .map(|f| {
+                        if f.data_type() == &DataType::Utf8 {
+                            Arc::new(Field::new(f.name(), DataType::Utf8View, f.is_nullable()))
+                        } else {
+                            f.clone()
+                        }
+                    })
+                    .collect::<Vec<_>>();
+                let new_schema =
+                    Schema::new(fields).with_metadata(schema.schema().metadata().clone());
+                final_schemas.insert(stream.clone(), Arc::new(SchemaCache::new(new_schema)));
+            }
+            final_schemas
+        } else {
+            used_schemas.clone()
+        };
+
         let is_complex = is_complex_query(&mut statement);
+
         Ok(Sql {
             sql: statement.to_string(),
             is_complex,
@@ -339,7 +365,7 @@ impl Sql {
             prefix_items: prefix_column_visitor.prefix_items,
             columns,
             aliases,
-            schemas: used_schemas,
+            schemas: final_schemas,
             limit,
             offset,
             time_range: Some((query.start_time, query.end_time)),