diff --git a/src/config/src/config.rs b/src/config/src/config.rs
index f053116b3c..f94e64b89c 100644
--- a/src/config/src/config.rs
+++ b/src/config/src/config.rs
@@ -512,6 +512,9 @@ pub struct Common {
     pub wal_line_mode_enabled: bool,
     #[env_config(name = "ZO_COLUMN_TIMESTAMP", default = "_timestamp")]
     pub column_timestamp: String,
+    // TODO: should rename to column_all
+    #[env_config(name = "ZO_CONCATENATED_SCHEMA_FIELD_NAME", default = "_all")]
+    pub column_all: String,
     #[env_config(name = "ZO_WIDENING_SCHEMA_EVOLUTION", default = true)]
     pub widening_schema_evolution: bool,
     #[env_config(name = "ZO_SKIP_SCHEMA_VALIDATION", default = false)]
@@ -530,6 +533,8 @@ pub struct Common {
     pub feature_query_partition_strategy: String,
     #[env_config(name = "ZO_FEATURE_QUERY_INFER_SCHEMA", default = false)]
     pub feature_query_infer_schema: bool,
+    #[env_config(name = "ZO_FEATURE_QUERY_EXCLUDE_ALL", default = true)]
+    pub feature_query_exclude_all: bool,
     #[env_config(name = "ZO_UI_ENABLED", default = true)]
     pub ui_enabled: bool,
     #[env_config(name = "ZO_UI_SQL_BASE64_ENABLED", default = false)]
@@ -678,8 +683,6 @@ pub struct Common {
     pub report_server_url: String,
     #[env_config(name = "ZO_REPORT_SERVER_SKIP_TLS_VERIFY", default = false)]
     pub report_server_skip_tls_verify: bool,
-    #[env_config(name = "ZO_CONCATENATED_SCHEMA_FIELD_NAME", default = "_all")]
-    pub all_fields_name: String,
     #[env_config(name = "ZO_SCHEMA_CACHE_COMPRESS_ENABLED", default = false)]
     pub schema_cache_compress_enabled: bool,
     #[env_config(name = "ZO_SKIP_FORMAT_BULK_STREAM_NAME", default = false)]
diff --git a/src/handler/http/request/status/mod.rs b/src/handler/http/request/status/mod.rs
index 061d0bd9dd..d0be77d6bc 100644
--- a/src/handler/http/request/status/mod.rs
+++ b/src/handler/http/request/status/mod.rs
@@ -267,7 +267,7 @@ pub async fn zo_config() -> Result<HttpResponse, Error> {
         meta_org: cfg.common.usage_org.to_string(),
         quick_mode_enabled: cfg.limit.quick_mode_enabled,
         user_defined_schemas_enabled: cfg.common.allow_user_defined_schemas,
-        all_fields_name: cfg.common.all_fields_name.to_string(),
+        all_fields_name: cfg.common.column_all.to_string(),
     }))
 }
 
diff --git a/src/service/compact/file_list_deleted.rs b/src/service/compact/file_list_deleted.rs
index ce485293de..c5a0d73d77 100644
--- a/src/service/compact/file_list_deleted.rs
+++ b/src/service/compact/file_list_deleted.rs
@@ -58,7 +58,7 @@ pub async fn delete(
             if file.1 {
                 Some(format!(
                     "files{}/{}",
-                    config::get_config().common.all_fields_name,
+                    config::get_config().common.column_all,
                     file.0.strip_prefix("files/").unwrap()
                 ))
             } else {
diff --git a/src/service/compact/flatten.rs b/src/service/compact/flatten.rs
index b8df816d89..d59d11012f 100644
--- a/src/service/compact/flatten.rs
+++ b/src/service/compact/flatten.rs
@@ -188,7 +188,7 @@ pub async fn generate_file(file: &FileKey) -> Result<(), anyhow::Error> {
     let full_text_search_fields = stream_setting.full_text_search_keys;
     let new_file = format!(
         "files{}/{}",
-        get_config().common.all_fields_name,
+        get_config().common.column_all,
         file.key.strip_prefix("files/").unwrap()
     );
     let new_schema = new_batches.first().unwrap().schema();
@@ -227,7 +227,7 @@ fn generate_vertical_partition_recordbatch(
     if records_len == 0 {
         return Ok(Vec::new());
     }
-    let Ok(all_field_idx) = schema.index_of(&get_config().common.all_fields_name) else {
+    let Ok(all_field_idx) = schema.index_of(&get_config().common.column_all) else {
         return Ok(vec![batches]);
     };
 
diff --git a/src/service/logs/mod.rs b/src/service/logs/mod.rs
index b087d262bc..7b95e858ae 100644
--- a/src/service/logs/mod.rs
+++ b/src/service/logs/mod.rs
@@ -476,7 +476,7 @@ pub fn refactor_map(
 
     if has_elements {
         new_map.insert(
-            get_config().common.all_fields_name.to_string(),
+            get_config().common.column_all.to_string(),
             Value::String(String::from_utf8(non_schema_map).unwrap()),
         );
     }
diff --git a/src/service/schema.rs b/src/service/schema.rs
index 1c9bfd0757..74b36e8d39 100644
--- a/src/service/schema.rs
+++ b/src/service/schema.rs
@@ -317,8 +317,8 @@ pub fn generate_schema_for_defined_schema_fields(
     if !fields.contains(&cfg.common.column_timestamp) {
         fields.insert(&cfg.common.column_timestamp);
     }
-    if !fields.contains(&cfg.common.all_fields_name) {
-        fields.insert(&cfg.common.all_fields_name);
+    if !fields.contains(&cfg.common.column_all) {
+        fields.insert(&cfg.common.column_all);
     }
     let mut new_fields = Vec::with_capacity(fields.len());
     for field in fields {
diff --git a/src/service/search/grpc/mod.rs b/src/service/search/grpc/mod.rs
index bec6091529..4fc4d696ff 100644
--- a/src/service/search/grpc/mod.rs
+++ b/src/service/search/grpc/mod.rs
@@ -430,8 +430,8 @@ fn generate_select_start_search_schema(
         if !fields.contains(&cfg.common.column_timestamp) {
             fields.insert(cfg.common.column_timestamp.to_string());
         }
-        if !fields.contains(&cfg.common.all_fields_name) {
-            fields.insert(cfg.common.all_fields_name.to_string());
+        if !cfg.common.feature_query_exclude_all && !fields.contains(&cfg.common.column_all) {
+            fields.insert(cfg.common.column_all.to_string());
         }
         let new_fields = fields
             .iter()