diff --git a/Cargo.lock b/Cargo.lock index 054ad8f4cb..6345413ddb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5427,7 +5427,7 @@ checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" [[package]] name = "openobserve" -version = "0.10.7" +version = "0.10.8" dependencies = [ "actix-cors", "actix-http", diff --git a/Cargo.toml b/Cargo.toml index 199108953a..c9c6f17a7d 100644 --- a/Cargo.toml +++ b/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 diff --git a/src/infra/src/schema/mod.rs b/src/infra/src/schema/mod.rs index 9105afd6c6..f05628aed8 100644 --- a/src/infra/src/schema/mod.rs +++ b/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(), diff --git a/src/job/files/parquet.rs b/src/job/files/parquet.rs index c33cac42dc..5843169fb8 100644 --- a/src/job/files/parquet.rs +++ b/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();