feat: set bloom filter for index stream (#3739)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

- **New Features**
- Enhanced schema handling logic to ensure correct processing based on
metadata.
- Added bloom filter settings for specific fields in schema creation and
updates.

- **Bug Fixes**
- Improved error handling in schema index generation to manage missing
schemas effectively.

- **Chores**
	- Updated version from "0.10.7" to "0.10.8".

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Cargo.lock

@@ -5427,7 +5427,7 @@ checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381"
 
 [[package]]
 name = "openobserve"
-version = "0.10.7"
+version = "0.10.8"
 dependencies = [
  "actix-cors",
  "actix-http",

Cargo.toml

@@ -17,7 +17,7 @@ keywords = [
 license = "AGPL-3.0-only"
 name = "openobserve"
 repository = "https://github.com/openobserve/openobserve/"
-version = "0.10.7"
+version = "0.10.8"
 publish = false
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 

src/infra/src/schema/mod.rs

@@ -307,8 +307,7 @@ pub async fn merge(
                         json::to_vec(&vec![{
                             // there is no schema, just set the new schema
                             let schema_metadata = inferred_schema.metadata();
-                            tx.send(Some((inferred_schema.clone(), vec![]))).unwrap();
-                            if schema_metadata.contains_key("created_at")
+                            let inferred_schema = if schema_metadata.contains_key("created_at")
                                 && schema_metadata.contains_key("start_dt")
                             {
                                 inferred_schema
@@ -325,7 +324,9 @@ pub async fn merge(
                                         .insert("start_dt".to_string(), start_dt.to_string());
                                 }
                                 inferred_schema.with_metadata(schema_metadata)
-                            }
+                            };
+                            tx.send(Some((inferred_schema.clone(), vec![]))).unwrap();
+                            inferred_schema
                         }])
                         .unwrap()
                         .into(),

src/job/files/parquet.rs

@@ -24,7 +24,7 @@ use bytes::Bytes;
 use chrono::{Duration, Utc};
 use config::{
     cluster, get_config,
-    meta::stream::{FileKey, FileMeta, PartitionTimeLevel, StreamType},
+    meta::stream::{FileKey, FileMeta, PartitionTimeLevel, StreamSettings, StreamType},
     metrics,
     utils::{
         arrow::record_batches_to_json_rows,
@@ -683,15 +683,28 @@ pub(crate) async fn generate_index_on_ingester(
     .await;
 
     if !schema_chk.has_fields {
-        db::schema::merge(
+        // create schema
+        let Some((schema, _)) = db::schema::merge(
             org_id,
             stream_name,
             StreamType::Index,
             idx_schema.as_ref(),
             Some(Utc::now().timestamp_micros()),
         )
-        .await
-        .unwrap();
+        .await?
+        else {
+            return Err(anyhow::anyhow!(
+                "generate_index_on_ingester error: schema not found"
+            ));
+        };
+        // update schema to enable bloomfilter for field: term, file_name
+        let settings = StreamSettings {
+            bloom_filter_fields: vec!["term".to_string(), "file_name".to_string()],
+            ..Default::default()
+        };
+        let mut metadata = schema.metadata().clone();
+        metadata.insert("settings".to_string(), json::to_string(&settings).unwrap());
+        db::schema::update_setting(org_id, stream_name, StreamType::Index, metadata).await?;
     }
     let schema_key = idx_schema.hash_key();
     let schema_key_str = schema_key.as_str();