diff --git a/src/api/management/src/models/alerts/requests.rs b/src/api/management/src/models/alerts/requests.rs index ccae472276..3e1deebb56 100644 --- a/src/api/management/src/models/alerts/requests.rs +++ b/src/api/management/src/models/alerts/requests.rs @@ -119,16 +119,6 @@ impl CreateAlertRequestBody { } } -/// Documentation union for the unified create endpoint. Runtime dispatch is -/// still driven by `alert_type`; this wrapper makes the OpenAPI contract -/// explicit for clients that generate discriminated request types. -#[derive(Clone, Debug, Deserialize, ToSchema)] -#[serde(untagged)] -pub enum CreateAlertRequestSchema { - Query(CreateAlertRequestBody), - Composite(CreateAlertRequestBody), -} - /// Anomaly-detection-specific fields for `CreateAlertRequestBody`. #[derive(Clone, Debug, Deserialize, Serialize, ToSchema)] pub struct AnomalyAlertFields { diff --git a/src/api/management/src/request/alerts/mod.rs b/src/api/management/src/request/alerts/mod.rs index 7bb8fb9298..3e957861e2 100644 --- a/src/api/management/src/request/alerts/mod.rs +++ b/src/api/management/src/request/alerts/mod.rs @@ -54,8 +54,8 @@ use crate::{ models::alerts::{ requests::{ AlertBulkEnableRequest, CloneAlertRequestBody, CreateAlertRequestBody, - CreateAlertRequestSchema, EnableAlertQuery, GenerateSqlRequestBody, ListAlertsQuery, - MoveAlertsRequestBody, UpdateAlertRequestBody, ValidateCompositeRequestBody, + EnableAlertQuery, GenerateSqlRequestBody, ListAlertsQuery, MoveAlertsRequestBody, + UpdateAlertRequestBody, ValidateCompositeRequestBody, }, responses::{ AlertBulkEnableResponse, AlertGroupLabel, AlertGroupResponseItem, @@ -98,7 +98,7 @@ pub mod templates; ("org_id" = String, Path, description = "Organization name"), ("folder" = Option, Query, description = "Folder ID (Required if alert folder is not the default folder)"), ), - request_body(content = inline(CreateAlertRequestSchema), description = "Alert data", content_type = "application/json"), + request_body(content = inline(CreateAlertRequestBody), description = "Alert data", content_type = "application/json"), responses( (status = 200, description = "Success", content_type = "application/json", body = Object), (status = 400, description = "Error", content_type = "application/json", body = ()), diff --git a/src/api/search/src/search/mod.rs b/src/api/search/src/search/mod.rs index 2e770b6084..14ebdaa0f7 100644 --- a/src/api/search/src/search/mod.rs +++ b/src/api/search/src/search/mod.rs @@ -24,7 +24,7 @@ use axum::{ }; use chrono::Utc; use config::{ - DISTINCT_FIELDS, META_ORG_ID, TIMESTAMP_COL_NAME, get_config, + META_ORG_ID, TIMESTAMP_COL_NAME, get_config, meta::{ search::{ AgentSearchMode, Request, ResultSchemaResponse, SearchEventType, @@ -35,7 +35,7 @@ use config::{ sql::resolve_stream_names, stream::StreamType, }, - utils::{base64, json, time::now_micros, util::DISTINCT_STREAM_PREFIX}, + utils::{base64, json, time::now_micros}, }; use db::enrichment_table; use error_utils::map_error_to_http_response; @@ -84,80 +84,6 @@ pub mod search_job; pub mod search_stream; pub mod utils; -async fn can_use_distinct_stream( - org_id: &str, - stream_name: &str, - stream_type: StreamType, - fields: &[String], - query: &config::meta::search::Query, - start_time: i64, -) -> bool { - if !matches!(stream_type, StreamType::Logs | StreamType::Traces) { - return false; - } - - let stream_settings = infra::schema::get_settings(org_id, stream_name, stream_type) - .await - .unwrap_or_default(); - - if !stream_settings.enable_distinct_fields { - return false; - } - - // all fields which are requested must be in the distinct stream - let all_fields_distinct = fields.iter().all(|f| { - if DISTINCT_FIELDS.contains(f) { - return true; - } - if f == "count" { - // count is reserved field from oo side, so if user has count field - // in original stream, it won't actually be in the distinct stream, so - // we need to fallback to normal search - return false; - } - stream_settings - .distinct_value_fields - .iter() - .any(|entry| entry.name == *f && entry.added_ts <= start_time) - }); - - // all the fields used in the query sent must be in the distinct stream - #[allow(deprecated)] - let query_fields: Vec = - match search::sql::Sql::new(&(query.clone().into()), org_id, stream_type, None).await { - // if sql is invalid, we let it follow the original search and fail - Err(_) => return false, - Ok(sql) => { - // check if sql contains any filters from which field cannot be inferred. - // where clause can contain match_all and a valid field which is in distinct stream - // but since there is match_all, we cannot infer the field from the where clause - // so we need to return false - if sql.has_match_all { - return false; - } - sql.columns.values().flatten().cloned().collect() - } - }; - - let all_query_fields_distinct = query_fields.iter().all(|f| { - if DISTINCT_FIELDS.contains(f) { - return true; - } - if f == "count" { - // count is reserved field from oo side, so if user has count field - // in original stream, it won't actually be in the distinct stream, so - // we need to fallback to normal search - return false; - } - stream_settings - .distinct_value_fields - .iter() - .any(|entry| entry.name == *f && entry.added_ts <= start_time) - }); - - all_fields_distinct && all_query_fields_distinct -} - /// SearchStreamData #[utoipa::path( @@ -869,13 +795,7 @@ pub async fn values( }; let trace_id = get_or_create_trace_id(&headers, &http_span); - // originally there was v1 which would to a full stream search - // and v2 which would do search on a distinct values stream iff - // the queried fields configured accordingly. - // Now we simply check if the fields in query are in the distinct stream or not, - // and change the search stream to the distinct stream, so we don't need any separate - // v2 fucntion. - values_v1( + values_inner( &org_id, stream_type, &stream_name, @@ -891,21 +811,10 @@ pub type FieldName = String; /// Builds a search request per field /// -/// This function builds a search request per field based on the given request and parameters. -/// Search request can basically be of two types: -/// 1. Search on distinct values stream -/// 2. Search on original data stream -/// If the field is a distinct field, we will search of the distinct values stream. -/// Otherwise, we will search on the original data stream. +/// The search runs on the original data stream; secondary index fields are +/// served by the tantivy TopN/Distinct optimization. /// -/// The `use_result_cache` parameter is used to determine the projection of the SQL query. -/// This flag will toggle the resultant requests between streaming aggregations and result cache. -/// By default, this function will produce an SQL which utilizes `Streaming Aggregations` to send -/// the results to the client. The SQL will be a simple aggregation query in case streaming -/// aggregations are used. If `use_result_cache` is set to true, the SQL projection will include a -/// histogram which will allow the use of result cache. -/// -/// Another parameter is `no_count` which is used to determine if the count is needed or not. +/// The `no_count` parameter is used to determine if the count is needed or not. /// `no_count` is used when only distinct values (sorted in alphabetical order) are needed but not /// the frequency of the values. For example, Dashboards, where we show the values listed in /// alphabetical order. @@ -990,7 +899,7 @@ pub async fn build_search_request_per_field( ..Default::default() }; - let (sql_where, can_use_distinct_stream) = match req.filter.as_ref() { + let sql_where = match req.filter.as_ref() { None => { if !decoded_sql.is_empty() { query.uses_zo_fn = functions::get_all_transform_keys(org_id) @@ -999,54 +908,27 @@ pub async fn build_search_request_per_field( .any(|fn_name| decoded_sql.contains(&format!("{fn_name}("))); // pick up where clause from sql - let sql_where_from_query = match pickup_where(&decoded_sql) { + match pickup_where(&decoded_sql) { Ok(Some(v)) => format!("WHERE {v}"), Ok(None) => "".to_string(), Err(e) => { return Err(Error::other(e)); } - }; - let can_use_distinct_stream = can_use_distinct_stream( - org_id, - stream_name, - stream_type, - &fields, - &query, - start_time, - ) - .await; - (sql_where_from_query, can_use_distinct_stream) + } } else { - ("".to_string(), false) + "".to_string() } } Some(v) => { if v.is_empty() { - ("".to_string(), false) + "".to_string() } else { let columns = v.splitn(2, '=').collect::>(); if columns.len() < 2 { return Err(Error::other("Invalid filter format")); } let vals = columns[1].split(',').collect::>().join("','"); - let sql_where = format!("WHERE {} IN ('{vals}')", columns[0]); - - // Define the default_sql here - let default_sql = format!("SELECT {TIMESTAMP_COL_NAME} FROM \"{stream_name}\""); - - query.sql = format!("{default_sql} {sql_where}"); - - let can_use_distinct_stream = can_use_distinct_stream( - org_id, - stream_name, - stream_type, - &fields, - &query, - start_time, - ) - .await; - - (sql_where, can_use_distinct_stream) + format!("WHERE {} IN ('{vals}')", columns[0]) } } }; @@ -1067,24 +949,6 @@ pub async fn build_search_request_per_field( agent_options: None, }; - let distinct_prefix = if can_use_distinct_stream { - format!("{}{}_", DISTINCT_STREAM_PREFIX, stream_type.as_str()) - } else { - "".to_string() - }; - - let count_fn = if can_use_distinct_stream { - "SUM(count)".to_string() - } else { - "COUNT(*)".to_string() - }; - - let actual_stream_type = if can_use_distinct_stream { - StreamType::Metadata - } else { - stream_type - }; - let size = req.query.size; let mut requests = Vec::new(); for field in fields { @@ -1097,11 +961,11 @@ pub async fn build_search_request_per_field( }; let sql = if no_count { format!( - "SELECT \"{field}\" AS zo_sql_key FROM \"{distinct_prefix}{stream_name}\" {sql_where} GROUP BY zo_sql_key order by zo_sql_key asc limit {size}" + "SELECT \"{field}\" AS zo_sql_key FROM \"{stream_name}\" {sql_where} GROUP BY zo_sql_key order by zo_sql_key asc limit {size}" ) } else { format!( - "SELECT \"{field}\" AS zo_sql_key, {count_fn} AS zo_sql_num FROM \"{distinct_prefix}{stream_name}\" {sql_where} GROUP BY zo_sql_key order by zo_sql_num desc limit {size}" + "SELECT \"{field}\" AS zo_sql_key, COUNT(*) AS zo_sql_num FROM \"{stream_name}\" {sql_where} GROUP BY zo_sql_key order by zo_sql_num desc limit {size}" ) }; @@ -1109,17 +973,15 @@ pub async fn build_search_request_per_field( req.query.size = size; req.query.from = 0; req.query.sql = sql; - requests.push((req, actual_stream_type, field)); + requests.push((req, stream_type, field)); } Ok(requests) } -// If all fields requested in the query AND fields from the -// sql query in the query are stored in distinct stream, -// this will search on the distinct stream, otherwise -// just search on the original data -async fn values_v1( +// search values on the original data stream; secondary index fields are +// served by the tantivy TopN/Distinct optimization +async fn values_inner( org_id: &str, stream_type: StreamType, stream_name: &str, @@ -1280,17 +1142,6 @@ async fn values_v1( query_fn: query_fn.clone(), ..Default::default() }; - // check if we can use the distinct stream for this query - let use_distinct_stream = can_use_distinct_stream( - org_id, - stream_name, - stream_type, - &fields, - &req_query, - start_time, - ) - .await; - let mut req = config::meta::search::Request { query: req_query, encoding: config::meta::search::RequestEncoding::Empty, @@ -1332,31 +1183,13 @@ async fn values_v1( sql_where.clone() }; - let distinct_prefix; - let count_fn; - let actual_stream_type; - - if use_distinct_stream { - distinct_prefix = format!("{}{}_", DISTINCT_STREAM_PREFIX, stream_type.as_str()); - // if we are using distinct stream, we have already partially aggregated - // the counts, so we need to sum over that field - count_fn = "SUM(count)"; - // distinct_values_* stream is metadata - actual_stream_type = StreamType::Metadata; - } else { - distinct_prefix = "".to_owned(); - // for non-distinct fields, we need the actual count - count_fn = "COUNT(*)"; - actual_stream_type = stream_type; - } - let sql = if no_count { format!( - "SELECT \"{field}\" AS zo_sql_key FROM \"{distinct_prefix}{stream_name}\" {sql_where} GROUP BY zo_sql_key ORDER BY zo_sql_key ASC" + "SELECT \"{field}\" AS zo_sql_key FROM \"{stream_name}\" {sql_where} GROUP BY zo_sql_key ORDER BY zo_sql_key ASC" ) } else { format!( - "SELECT \"{field}\" AS zo_sql_key, {count_fn} AS zo_sql_num FROM \"{distinct_prefix}{stream_name}\" {sql_where} GROUP BY zo_sql_key ORDER BY zo_sql_num DESC, zo_sql_key ASC" + "SELECT \"{field}\" AS zo_sql_key, COUNT(*) AS zo_sql_num FROM \"{stream_name}\" {sql_where} GROUP BY zo_sql_key ORDER BY zo_sql_num DESC, zo_sql_key ASC" ) }; let mut req = req.clone(); @@ -1370,7 +1203,7 @@ async fn values_v1( let search_res = SearchService::cache::search( &trace_id, org_id, - actual_stream_type, + stream_type, Some(user_id.to_string()), &req, "".to_string(), diff --git a/src/config/src/config.rs b/src/config/src/config.rs index afb8842dfd..83e07a4318 100644 --- a/src/config/src/config.rs +++ b/src/config/src/config.rs @@ -217,18 +217,18 @@ pub static QUICK_MODEL_FIELDS: Lazy> = Lazy::new(|| { fields }); -const _DEFAULT_DISTINCT_FIELDS: [&str; 2] = ["service_name", "operation_name"]; -pub static DISTINCT_FIELDS: Lazy> = Lazy::new(|| { +const _DEFAULT_BLOOM_FILTER_FIELDS: [&str; 1] = ["trace_id"]; +pub static BLOOM_FILTER_DEFAULT_FIELDS: Lazy> = Lazy::new(|| { let cfg = get_config(); let default_fields: &[&str] = if cfg.common.feature_default_index_fields_enabled { - &_DEFAULT_DISTINCT_FIELDS + &_DEFAULT_BLOOM_FILTER_FIELDS } else { &[] }; let mut fields = chain( default_fields.iter().map(|s| s.to_string()), cfg.common - .feature_distinct_extra_fields + .feature_bloom_filter_extra_fields .split(',') .filter_map(|s| { let s = s.trim(); @@ -245,25 +245,6 @@ pub static DISTINCT_FIELDS: Lazy> = Lazy::new(|| { fields }); -pub static BLOOM_FILTER_DEFAULT_FIELDS: Lazy> = Lazy::new(|| { - let mut fields = get_config() - .common - .feature_bloom_filter_extra_fields - .split(',') - .filter_map(|s| { - let s = s.trim(); - if s.is_empty() { - None - } else { - Some(s.to_string()) - } - }) - .collect::>(); - fields.sort(); - fields.dedup(); - fields -}); - const _DEFAULT_SEARCH_AROUND_FIELDS: [&str; 6] = [ "k8s_cluster", "k8s_namespace_name", @@ -1397,7 +1378,7 @@ pub struct Common { #[env_config( name = "ZO_FEATURE_DEFAULT_INDEX_FIELDS_ENABLED", default = true, - help = "When false, the built-in default fields for full text search, secondary index and distinct values are disabled; only the fields from the *_EXTRA_FIELDS ENVs and per-stream settings are used" + help = "When false, the built-in default fields for full text search and secondary index are disabled; only the fields from the *_EXTRA_FIELDS ENVs and per-stream settings are used" )] pub feature_default_index_fields_enabled: bool, #[env_config(name = "ZO_FEATURE_FULLTEXT_EXTRA_FIELDS", default = "")] @@ -1410,8 +1391,6 @@ pub struct Common { help = "Comma-separated fields to build bloom filter on for all streams, replaces the deprecated ZO_BLOOM_FILTER_DEFAULT_FIELDS" )] pub feature_bloom_filter_extra_fields: String, - #[env_config(name = "ZO_FEATURE_DISTINCT_EXTRA_FIELDS", default = "")] - pub feature_distinct_extra_fields: String, #[env_config(name = "ZO_FEATURE_QUICK_MODE_FIELDS", default = "")] pub feature_quick_mode_fields: String, #[env_config(name = "ZO_FEATURE_QUERY_QUEUE_ENABLED", default = true)] @@ -2271,10 +2250,6 @@ pub struct Limit { help = "max time of transaction will retry" )] pub meta_transaction_retries: usize, - #[env_config(name = "ZO_DISTINCT_VALUES_INTERVAL", default = 10)] // seconds - pub distinct_values_interval: u64, - #[env_config(name = "ZO_DISTINCT_VALUES_HOURLY", default = false)] - pub distinct_values_hourly: bool, #[env_config(name = "ZO_CONSISTENT_HASH_VNODES", default = 1000)] pub consistent_hash_vnodes: usize, #[env_config( diff --git a/src/core/src/lib.rs b/src/core/src/lib.rs index 45b035e2a1..84f9a3a096 100644 --- a/src/core/src/lib.rs +++ b/src/core/src/lib.rs @@ -40,7 +40,6 @@ pub mod kv; #[cfg(feature = "enterprise")] pub mod llm_evaluations; pub mod logs; -pub mod metadata; pub mod metrics; #[cfg(feature = "enterprise")] pub mod ofga; diff --git a/src/core/src/logs/mod.rs b/src/core/src/logs/mod.rs index f312cdb9b3..7bb0a85391 100644 --- a/src/core/src/logs/mod.rs +++ b/src/core/src/logs/mod.rs @@ -24,7 +24,7 @@ use ::stream::get_stream; use arrow_schema::{DataType, Field}; use bulk::SCHEMA_CONFORMANCE_FAILED; use config::{ - DISTINCT_FIELDS, META_ORG_ID, SIZE_IN_MB, get_config, + META_ORG_ID, SIZE_IN_MB, get_config, meta::{ alerts::alert::Alert, self_reporting::usage::{RequestStats, UsageType}, @@ -35,7 +35,6 @@ use config::{ json::{Map, Value, estimate_json_bytes, get_string_value}, schema_ext::SchemaExt, time::now_micros, - util::DISTINCT_STREAM_PREFIX, }, }; use db; @@ -50,7 +49,6 @@ use crate::{ alerts::alert::AlertExt, common::meta::stream::SchemaRecords, ingestion::{TriggerAlertData, evaluate_trigger, get_write_partition_key, write_file}, - metadata::{MetadataItem, MetadataType, distinct_values::DvItem, write}, }; pub mod bulk; @@ -381,8 +379,6 @@ async fn write_logs( None => Arc::new(latest_schema), }; - let mut distinct_values = Vec::with_capacity(16); - let mut write_buf: HashMap = HashMap::new(); for (timestamp, mut record_val) in json_data { @@ -490,30 +486,6 @@ async fn write_logs( } // end check for alert triggers - // get distinct_value items - if stream_settings.enable_distinct_fields { - let mut map = Map::new(); - for field in DISTINCT_FIELDS.iter().chain( - stream_settings - .distinct_value_fields - .iter() - .map(|f| &f.name), - ) { - if let Some(val) = record_val.get(field) { - map.insert(field.clone(), val.clone()); - } - } - - if !map.is_empty() { - // add distinct values - distinct_values.push(MetadataItem::DistinctValues(DvItem { - stream_type: StreamType::Logs, - stream_name: stream_name.to_string(), - value: map, - })); - } - } - // get hour key let hour_key = get_write_partition_key( timestamp, @@ -565,15 +537,6 @@ async fn write_logs( ) .await?; - // send distinct_values - if !distinct_values.is_empty() - && !stream_name.starts_with(DISTINCT_STREAM_PREFIX) - && stream_settings.enable_distinct_fields - && let Err(e) = write(org_id, MetadataType::DistinctValues, distinct_values).await - { - log::error!("Error while writing distinct values: {e}"); - } - // only one trigger per request if !triggers.is_empty() { tokio::spawn(async move { evaluate_trigger(triggers).await }); diff --git a/src/core/src/metadata/distinct_values.rs b/src/core/src/metadata/distinct_values.rs deleted file mode 100644 index 0218bbe286..0000000000 --- a/src/core/src/metadata/distinct_values.rs +++ /dev/null @@ -1,417 +0,0 @@ -// Copyright 2026 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 . - -use std::{ - collections::HashMap, - sync::{ - Arc, LazyLock as Lazy, - atomic::{AtomicBool, Ordering}, - }, -}; - -use arrow_schema::{DataType, Field, Schema}; -use config::{ - FxIndexMap, TIMESTAMP_COL_NAME, get_config, - meta::stream::StreamType, - spawn_pausable_job, - utils::{ - json, schema::infer_json_schema_from_map, time::now_micros, util::get_distinct_stream_name, - }, -}; -use db; -use infra::{ - errors::{Error, Result}, - schema::{SchemaCache, get_partition_time_level}, -}; -use schema::get_schema_changes; -use serde::{Deserialize, Serialize}; -use serde_json::{Map, Value}; -use tokio::sync::{RwLock, mpsc}; - -use crate::{ - common::meta::stream::SchemaRecords, - ingestion::{self, get_thread_id}, - metadata::{Metadata, MetadataItem}, -}; - -const CHANNEL_SIZE: usize = 10240; - -pub(crate) static INSTANCE: Lazy = Lazy::new(DistinctValues::new); - -type MemTable = FxIndexMap>; - -pub struct DistinctValues { - channel: Arc>, - shutdown: Arc, - mem_table: Arc>, -} - -#[derive(Debug, Default, Eq, Hash, PartialEq, Clone, Serialize, Deserialize)] -pub struct DvItem { - pub stream_type: StreamType, - pub stream_name: String, - pub value: Map, -} - -#[derive(Debug)] -enum DvEventType { - Add, - Shutudown, -} - -#[derive(Debug)] -struct DvEvent { - org_id: String, - item: DvItem, - count: u32, - ev_type: DvEventType, -} - -impl DvEvent { - pub fn new(org_id: &str, item: DvItem, count: u32) -> Self { - Self { - org_id: org_id.to_string(), - item, - count, - ev_type: DvEventType::Add, - } - } - pub fn shutdown() -> Self { - Self { - org_id: String::from(""), - item: DvItem::default(), - count: 0, - ev_type: DvEventType::Shutudown, - } - } -} - -impl Default for DistinctValues { - fn default() -> Self { - Self::new() - } -} - -impl DistinctValues { - pub fn new() -> Self { - spawn_pausable_job!( - "distinct_values_flush", - get_config().limit.distinct_values_interval, - { - if let Err(e) = INSTANCE.flush().await { - log::error!("[DISTINCT_VALUES] error flush data to wal: {e}"); - } - } - ); - Self { - channel: handle_channel(), - shutdown: Arc::new(AtomicBool::new(false)), - mem_table: Arc::new(RwLock::new(FxIndexMap::default())), - } - } -} - -fn handle_channel() -> Arc> { - let (tx, mut rx) = mpsc::channel::(CHANNEL_SIZE); - tokio::task::spawn(async move { - loop { - let event = match rx.recv().await { - Some(v) => v, - None => { - log::info!("[DISTINCT_VALUES] event channel closed"); - break; - } - }; - if let DvEventType::Shutudown = event.ev_type { - if let Err(e) = INSTANCE.flush().await { - log::error!("[DISTINCT_VALUES] flush error: {e}"); - } - INSTANCE.shutdown.store(true, Ordering::Release); - break; - } - let mut mem_table = INSTANCE.mem_table.write().await; - let entry = mem_table.entry(event.org_id).or_default(); - let field_entry = entry.entry(event.item).or_default(); - *field_entry += event.count; - } - log::info!("[DISTINCT_VALUES] event loop exit"); - }); - Arc::new(tx) -} - -impl Metadata for DistinctValues { - fn generate_schema(&self) -> Arc { - // distinct values will always have _timestamp and - // count, rest will be dynamically determined - Arc::new(Schema::new(vec![ - Field::new(TIMESTAMP_COL_NAME, DataType::Int64, false), - Field::new("count", DataType::Int64, true), - ])) - } - - async fn write(&self, org_id: &str, data: Vec) -> Result<()> { - let mut group_items: FxIndexMap = FxIndexMap::default(); - for item in data { - let MetadataItem::DistinctValues(mut item) = item; - // these two are reserved, so we remove them if present - item.value.remove("count"); - item.value.remove(TIMESTAMP_COL_NAME); - let count = group_items.entry(item).or_default(); - *count += 1; - } - for (item, count) in group_items { - self.channel - .send(DvEvent::new(org_id, item, count)) - .await - .map_err(|v| Error::Message(v.to_string()))?; - } - Ok(()) - } - - async fn flush(&self) -> Result<()> { - let cfg = get_config(); - let mut mem_table = self.mem_table.write().await; - let mut new_table: MemTable = FxIndexMap::default(); - std::mem::swap(&mut new_table, &mut *mem_table); - drop(mem_table); - - // write to wal - let timestamp = now_micros(); - let default_schema = self.generate_schema(); - - // transpose the table - let mut table: HashMap<_, Vec<(Map, u32)>> = HashMap::new(); - for (org_id, items) in new_table { - for (item, count) in items { - let key = (org_id.clone(), item.stream_name, item.stream_type); - let entry = table.entry(key).or_default(); - entry.push((item.value, count)); - } - } - - for ((org_id, stream_name, stream_type), items) in table { - if items.is_empty() { - continue; - } - - let distinct_stream_name = get_distinct_stream_name(stream_type, &stream_name); - - // check for schema - let mut db_schema = - infra::schema::get_cache(&org_id, &distinct_stream_name, StreamType::Metadata) - .await?; - let mut is_new = false; - if !db_schema.fields_map().contains_key(TIMESTAMP_COL_NAME) { - is_new = true; - let schema = default_schema.as_ref().clone(); - match db::schema::merge( - &org_id, - &distinct_stream_name, - StreamType::Metadata, - &schema, - Some(timestamp), - ) - .await - { - Ok(None) => {} - Ok(Some((s, _))) => db_schema = SchemaCache::new(s), - Err(e) => { - log::error!("[DISTINCT_VALUES] error while setting schema: {e}"); - return Err(Error::Message(e.to_string())); - } - }; - - if let Some(source_settings) = - infra::schema::get_settings(&org_id, &stream_name, stream_type).await - { - // local editable copy; persisted via save_stream_settings below - let mut new_settings = infra::schema::get_settings( - &org_id, - &distinct_stream_name, - StreamType::Metadata, - ) - .await - .map(|s| (*s).clone()) - .unwrap_or_default(); - new_settings.data_retention = source_settings.data_retention; - new_settings.extended_retention_days = - source_settings.extended_retention_days.clone(); - if let Err(e) = stream::save_stream_settings( - &org_id, - &distinct_stream_name, - StreamType::Metadata, - new_settings, - ) - .await - { - // in worse case the original and distinct stream will have different - // retention, but no point in failing the whole - // ingest operation of distinct values for that - log::warn!( - "error updating stream settings for retention for distinct values stream {org_id}/{distinct_stream_name} : {e}" - ); - } - } - } - - let inferred_schema = infer_json_schema_from_map( - &distinct_stream_name, - stream_type, - items.iter().map(|(v, _)| v), - )?; - let schema = if is_new || get_schema_changes(&db_schema, &inferred_schema).0 { - match db::schema::merge( - &org_id, - &distinct_stream_name, - StreamType::Metadata, - &inferred_schema, - Some(timestamp), - ) - .await - { - Ok(None) => db_schema.schema().clone(), - Ok(Some((s, _))) => Arc::new(s), - Err(e) => { - log::error!( - "[DISTINCT_VALUES] error while updating schema for {org_id}/{stream_name} : {e}" - ); - return Err(Error::Message(e.to_string())); - } - } - } else { - db_schema.schema().clone() - }; - let schema_key = db_schema.hash_key(); - - let mut buf: HashMap = HashMap::new(); - for (item, count) in items { - let mut data = json::to_value(item).unwrap(); - let data = data.as_object_mut().unwrap(); - data.insert("count".to_string(), json::Value::Number(count.into())); - data.insert( - TIMESTAMP_COL_NAME.to_string(), - json::Value::Number(timestamp.into()), - ); - let hour_key = ingestion::get_write_partition_key( - timestamp, - &vec![], - get_partition_time_level(StreamType::Metadata), - data, - Some(schema_key), - ); - let data = json::Value::Object(data.clone()); - let data_size = json::to_vec(&data).unwrap_or_default().len(); - - let hour_buf = buf.entry(hour_key).or_insert_with(|| SchemaRecords { - schema_key: schema_key.to_string(), - schema: schema.clone(), - records: vec![], - records_size: 0, - }); - hour_buf.records.push(Arc::new(data)); - hour_buf.records_size += data_size; - } - - let writer = ingester::get_writer( - get_thread_id(), - &org_id, - StreamType::Metadata.as_str(), - &distinct_stream_name, - ) - .await; - _ = ingestion::write_file( - &writer, - &org_id, - &distinct_stream_name, - buf, - !cfg.common.wal_fsync_disabled, - ) - .await; - - #[cfg(feature = "enterprise")] - { - use o2_openfga::{ - authorizer::authz::set_ownership_if_not_exists, - config::get_config as get_openfga_config, - }; - - // set ownership only in the first time - if is_new && get_openfga_config().enabled { - set_ownership_if_not_exists( - &org_id, - &format!("{}:{}", StreamType::Metadata, distinct_stream_name), - ) - .await; - } - } - } - Ok(()) - } - - async fn stop(&self) -> Result<()> { - let tx = self.channel.clone(); - tx.send(DvEvent::shutdown()) - .await - .map_err(|e| Error::Message(e.to_string()))?; - let mut i = 0; - while i < 10 { - if self.shutdown.load(Ordering::Relaxed) { - break; - } - tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; - log::info!("[DISTINCT_VALUES] shutting down"); - i += 1; - } - Ok(()) - } -} - -#[cfg(test)] -mod tests { - use config::meta::stream::StreamType; - - use super::*; - - fn make_item() -> DvItem { - DvItem { - stream_type: StreamType::Logs, - stream_name: "mystream".to_string(), - value: Map::new(), - } - } - - #[test] - fn test_dv_event_new_sets_org_and_count() { - let item = make_item(); - let ev = DvEvent::new("myorg", item.clone(), 5); - assert_eq!(ev.org_id, "myorg"); - assert_eq!(ev.count, 5); - assert_eq!(ev.item.stream_name, "mystream"); - } - - #[test] - fn test_dv_event_shutdown_has_empty_org() { - let ev = DvEvent::shutdown(); - assert!(ev.org_id.is_empty()); - assert_eq!(ev.count, 0); - } - - #[test] - fn test_dv_item_default_is_logs_type() { - let item = DvItem::default(); - assert_eq!(item.stream_type, StreamType::Logs); - assert!(item.stream_name.is_empty()); - } -} diff --git a/src/core/src/metadata/mod.rs b/src/core/src/metadata/mod.rs deleted file mode 100644 index 1dea95bdfd..0000000000 --- a/src/core/src/metadata/mod.rs +++ /dev/null @@ -1,106 +0,0 @@ -// Copyright 2026 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 . - -use std::{ - hash::Hash, - sync::{Arc, LazyLock as Lazy}, -}; - -use arrow_schema::Schema; -use serde::{Deserialize, Serialize}; - -use crate::metadata::distinct_values::DvItem; - -pub mod distinct_values; - -static METADATA_MANAGER: Lazy = Lazy::new(MetadataManager::new); - -#[derive(Debug, Eq, Hash, PartialEq, Clone, Serialize, Deserialize)] -pub enum MetadataItem { - DistinctValues(DvItem), -} - -pub enum MetadataType { - DistinctValues, -} - -pub struct MetadataManager {} - -pub trait Metadata { - fn generate_schema(&self) -> Arc; - fn write( - &self, - org_id: &str, - data: Vec, - ) -> impl std::future::Future> + Send; - fn flush(&self) -> impl std::future::Future> + Send; - fn stop(&self) -> impl std::future::Future> + Send; -} - -impl Default for MetadataManager { - fn default() -> Self { - Self::new() - } -} - -impl MetadataManager { - pub fn new() -> Self { - Self {} - } - - pub async fn close(&self) -> infra::errors::Result<()> { - if let Err(e) = distinct_values::INSTANCE.stop().await { - log::error!("[METADATA] error while closing: {e}"); - } - - Ok(()) - } -} - -pub async fn write( - org_id: &str, - mt: MetadataType, - data: Vec, -) -> infra::errors::Result<()> { - match mt { - MetadataType::DistinctValues => distinct_values::INSTANCE.write(org_id, data).await, - } -} - -pub async fn close() -> infra::errors::Result<()> { - // flush metadata - METADATA_MANAGER.close().await -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_metadata_manager_new() { - let _m = MetadataManager::new(); - } - - #[test] - fn test_metadata_manager_default() { - let _m = MetadataManager::default(); - } - - #[test] - fn test_metadata_type_variants() { - let t = MetadataType::DistinctValues; - assert!(matches!(t, MetadataType::DistinctValues)); - } -} diff --git a/src/core/src/traces/mod.rs b/src/core/src/traces/mod.rs index dc8b0b5e1a..9331ebfd13 100644 --- a/src/core/src/traces/mod.rs +++ b/src/core/src/traces/mod.rs @@ -29,7 +29,7 @@ use axum::{ use bytes::BytesMut; use chrono::{Duration, Utc}; use config::{ - DISTINCT_FIELDS, O2_INGEST_TS_COL_NAME, TIMESTAMP_COL_NAME, get_config, + O2_INGEST_TS_COL_NAME, TIMESTAMP_COL_NAME, get_config, meta::{ alerts::alert::Alert, gen_ai::GenAiAgentMappingConfig, @@ -38,7 +38,7 @@ use config::{ stream::{StreamParams, StreamPartition, StreamType}, }, metrics, - utils::{flatten, json, schema_ext::SchemaExt, time::now_micros, util::DISTINCT_STREAM_PREFIX}, + utils::{flatten, json, schema_ext::SchemaExt, time::now_micros}, }; use infra::schema::{SchemaCache, get_partition_time_level}; use ingestion_common::IngestUser; @@ -76,7 +76,6 @@ use crate::{ write_file, }, logs::O2IngestJsonData, - metadata::{MetadataItem, MetadataType, distinct_values::DvItem, write}, traces::otel::{OtelIngestionProcessor, is_llm_trace}, }; @@ -1419,10 +1418,6 @@ async fn write_traces( ) .await; - let stream_settings = infra::schema::get_settings(org_id, stream_name, StreamType::Traces) - .await - .unwrap_or_default(); - let mut partition_keys: Vec = vec![]; let partition_time_level = get_partition_time_level(StreamType::Traces); if stream_schema.has_partition_keys { @@ -1484,32 +1479,9 @@ async fn write_traces( } let mut data_buf: HashMap = HashMap::new(); - let mut distinct_values = Vec::with_capacity(16); // Start write data for (timestamp, record_val) in json_data { - // get distinct_value item - if stream_settings.enable_distinct_fields { - let mut map = Map::new(); - for field in DISTINCT_FIELDS.iter().chain( - stream_settings - .distinct_value_fields - .iter() - .map(|f| &f.name), - ) { - if let Some(val) = record_val.get(field) { - map.insert(field.clone(), val.clone()); - } - } - if !map.is_empty() { - distinct_values.push(MetadataItem::DistinctValues(DvItem { - stream_type: StreamType::Traces, - stream_name: stream_name.to_string(), - value: map, - })); - } - } - // Start check for alert trigger if let Some(alerts) = cur_stream_alerts && triggers.len() < alerts.len() @@ -1582,15 +1554,6 @@ async fn write_traces( std::io::Error::other(e.to_string()) })?; - // send distinct_values - if !distinct_values.is_empty() - && !stream_name.starts_with(DISTINCT_STREAM_PREFIX) - && stream_settings.enable_distinct_fields - && let Err(e) = write(org_id, MetadataType::DistinctValues, distinct_values).await - { - log::error!("Error while writing distinct values: {e}"); - } - // only one trigger per request evaluate_trigger(triggers).await; diff --git a/src/infra/src/db/tantivy_index.rs b/src/infra/src/db/tantivy_index.rs index 2af3c24d5f..506e7594f2 100644 --- a/src/infra/src/db/tantivy_index.rs +++ b/src/infra/src/db/tantivy_index.rs @@ -17,30 +17,52 @@ use tokio::sync::OnceCell; static TIMESTAMP_UPDATED_AT: OnceCell = OnceCell::const_new(); static SECONDARY_INDEX_UPDATED_AT: OnceCell = OnceCell::const_new(); +static DISTINCT_FIELDS_UPDATED_AT: OnceCell = OnceCell::const_new(); pub async fn get_ttv_timestamp_updated_at() -> i64 { *TIMESTAMP_UPDATED_AT - .get_or_init(|| get_or_create_updated_at("/tantivy/_timestamp/updated_at")) + .get_or_init(|| { + get_or_create_updated_at( + "/tantivy/_timestamp/updated_at", + config::utils::time::BASE_TIME.timestamp_micros(), + ) + }) .await } pub async fn get_ttv_secondary_index_updated_at() -> i64 { *SECONDARY_INDEX_UPDATED_AT - .get_or_init(|| get_or_create_updated_at("/tantivy/secondary_index/updated_at")) + .get_or_init(|| { + get_or_create_updated_at( + "/tantivy/secondary_index/updated_at", + config::utils::time::BASE_TIME.timestamp_micros(), + ) + }) .await } -async fn get_or_create_updated_at(key: &str) -> i64 { +/// when distinct value fields started being folded into the tantivy index +pub async fn get_ttv_distinct_fields_updated_at() -> i64 { + *DISTINCT_FIELDS_UPDATED_AT + .get_or_init(|| { + get_or_create_updated_at( + "/tantivy/distinct_fields/updated_at", + config::utils::time::now_micros(), + ) + }) + .await +} + +async fn get_or_create_updated_at(key: &str, default_ts: i64) -> i64 { let db = super::get_db().await; match db.get(key).await { Ok(ret) if !ret.is_empty() => String::from_utf8_lossy(&ret).parse::().unwrap(), _ => { - let timestamp = config::utils::time::BASE_TIME.timestamp_micros(); - let data = bytes::Bytes::from(timestamp.to_string()); + let data = bytes::Bytes::from(default_ts.to_string()); if let Err(err) = db.put(key, data, false, None).await { log::warn!("[infra::db::tantivy_index] failed to store {key}: {err}"); } - timestamp + default_ts } } } diff --git a/src/infra/src/schema/mod.rs b/src/infra/src/schema/mod.rs index 3d2939847f..53f7323a60 100644 --- a/src/infra/src/schema/mod.rs +++ b/src/infra/src/schema/mod.rs @@ -23,7 +23,7 @@ use chrono::Utc; use config::{ ALL_VALUES_COL_NAME, BLOOM_FILTER_DEFAULT_FIELDS, ORIGINAL_DATA_COL_NAME, RwAHashMap, RwHashMap, RwHashSet, SQL_FULL_TEXT_SEARCH_FIELDS, SQL_SECONDARY_INDEX_SEARCH_FIELDS, - get_config, + TIMESTAMP_COL_NAME, get_config, ider::SnowflakeIdGenerator, meta::stream::{PartitionTimeLevel, StreamSettings, StreamType}, stats::MemorySize, @@ -438,6 +438,31 @@ pub fn get_stream_setting_index_fields>( let mut fields = settings.index_fields.clone(); fields.extend(default_fields); fields.extend(settings.bloom_filter_fields.clone()); + // distinct value fields are secondary index fields while enabled; + // fields that cannot be indexed (fts / partition keys / reserved) are skipped + if settings.enable_distinct_fields && !settings.distinct_value_fields.is_empty() { + let cfg = get_config(); + let reserved = [ + TIMESTAMP_COL_NAME, + cfg.common.column_all.as_str(), + ALL_VALUES_COL_NAME, + ORIGINAL_DATA_COL_NAME, + ]; + fields.extend( + settings + .distinct_value_fields + .iter() + .map(|field| field.name.clone()) + .filter(|field| { + !settings.full_text_search_keys.contains(field) + && !settings + .partition_keys + .iter() + .any(|partition| partition.field == *field) + && !reserved.contains(&field.as_str()) + }), + ); + } fields } None => default_fields, @@ -468,6 +493,7 @@ pub fn get_stream_setting_index_updated_at_for_fields, created_at: Option, fields: &[String], + distinct_fields_updated_at: i64, ) -> i64 { if fields.is_empty() { return 0; @@ -490,12 +516,26 @@ pub fn get_stream_setting_index_updated_at_for_fields>().len(); + assert_eq!(unique_count, fields.len()); + + // distinct fields are not index fields when the feature is disabled + let settings = StreamSettings { + enable_distinct_fields: false, + ..settings + }; + let fields = get_stream_setting_index_fields(&Some(settings)); + assert!(fields.contains(&"index_field".to_string())); + assert!(!fields.contains(&"svc".to_string())); + } + + #[test] + fn test_get_stream_setting_index_updated_at_for_distinct_fields() { + use config::meta::stream::DistinctField; + + let created_at = 1_700_000_000_000_000; + let added_ts = 1_755_000_000_000_000; + let settings = StreamSettings { + index_updated_at: 1_750_000_000_000_000, + distinct_value_fields: vec![DistinctField { + name: "svc".to_string(), + added_ts, + }], + ..Default::default() + }; + + // a distinct-only field is indexed since its added_ts + let fields = vec!["svc".to_string()]; + let result = get_stream_setting_index_updated_at_for_fields( + &Some(settings.clone()), + Some(created_at), + &fields, + 0, + ); + assert_eq!(result, added_ts); + + // but never before the release-wide distinct fields cutoff + let cutoff = added_ts + 1_000_000; + let result = get_stream_setting_index_updated_at_for_fields( + &Some(settings.clone()), + Some(created_at), + &fields, + cutoff, + ); + assert_eq!(result, cutoff); + + // a field also present in index_fields keeps the index semantics + let settings_both = StreamSettings { + index_fields: vec!["svc".to_string()], + ..settings.clone() + }; + let result = get_stream_setting_index_updated_at_for_fields( + &Some(settings_both), + Some(created_at), + &fields, + 0, + ); + assert_eq!(result, 1_750_000_000_000_000); + + // disabled distinct fields fall back to the stream-level value + let settings_disabled = StreamSettings { + enable_distinct_fields: false, + ..settings + }; + let result = get_stream_setting_index_updated_at_for_fields( + &Some(settings_disabled), + Some(created_at), + &fields, + 0, + ); + assert_eq!(result, 1_750_000_000_000_000); + } + #[test] fn test_get_stream_setting_index_updated_at_for_fields() { let base_time = BASE_TIME.timestamp_micros(); @@ -1246,25 +1391,25 @@ mod tests { // no referenced fields -> no cutoff let fields = Vec::new(); let result = - get_stream_setting_index_updated_at_for_fields(&settings, Some(created_at), &fields); + get_stream_setting_index_updated_at_for_fields(&settings, Some(created_at), &fields, 0); assert_eq!(result, 0); // a field with its own entry uses it let fields = vec!["user_id".to_string()]; let result = - get_stream_setting_index_updated_at_for_fields(&settings, Some(created_at), &fields); + get_stream_setting_index_updated_at_for_fields(&settings, Some(created_at), &fields, 0); assert_eq!(result, 1_760_000_000_000_000); // a field without an entry falls back to the stream-level value let fields = vec!["level".to_string()]; let result = - get_stream_setting_index_updated_at_for_fields(&settings, Some(created_at), &fields); + get_stream_setting_index_updated_at_for_fields(&settings, Some(created_at), &fields, 0); assert_eq!(result, 1_750_000_000_000_000); // multiple fields use the max let fields = vec!["user_id".to_string(), "level".to_string()]; let result = - get_stream_setting_index_updated_at_for_fields(&settings, Some(created_at), &fields); + get_stream_setting_index_updated_at_for_fields(&settings, Some(created_at), &fields, 0); assert_eq!(result, 1_760_000_000_000_000); // None settings falls back to created_at @@ -1273,6 +1418,7 @@ mod tests { &None::, Some(created_at), &fields, + 0, ); assert_eq!(result, created_at); @@ -1283,20 +1429,21 @@ mod tests { let settings = Some(stream_settings); let fields = vec!["trace_id".to_string()]; let result = - get_stream_setting_index_updated_at_for_fields(&settings, Some(created_at), &fields); + get_stream_setting_index_updated_at_for_fields(&settings, Some(created_at), &fields, 0); assert_eq!(result, base_time); let fields = vec!["service_name".to_string()]; let result = get_stream_setting_index_updated_at_for_fields( &None::, Some(created_at), &fields, + 0, ); assert_eq!(result, base_time); // a default field combined with a configured field still uses the max let fields = vec!["trace_id".to_string(), "user_id".to_string()]; let result = - get_stream_setting_index_updated_at_for_fields(&settings, Some(created_at), &fields); + get_stream_setting_index_updated_at_for_fields(&settings, Some(created_at), &fields, 0); assert_eq!(result, 1_760_000_000_000_000); } diff --git a/src/main.rs b/src/main.rs index f2fcec10bc..29c592df03 100644 --- a/src/main.rs +++ b/src/main.rs @@ -35,7 +35,7 @@ use openobserve::{ telemetry::{enable_tracing, setup_logs}, }; use openobserve_api_http::handler::http::router::*; -use openobserve_core::{bootstrap, metadata}; +use openobserve_core::bootstrap; use openobserve_jobs::job; use tokio::sync::oneshot; #[cfg(feature = "enterprise")] @@ -232,8 +232,6 @@ async fn main() -> Result<(), anyhow::Error> { job_shutdown_rx.await.ok(); job_stopped_tx.send(()).ok(); - // flush distinct values - _ = metadata::close().await; // flush WAL cache to disk _ = ingester::flush_all().await; // flush compact offset cache to disk disk diff --git a/src/search/src/datafusion/merge/mod.rs b/src/search/src/datafusion/merge/mod.rs index ed64d8e8d1..944ea8a81f 100644 --- a/src/search/src/datafusion/merge/mod.rs +++ b/src/search/src/datafusion/merge/mod.rs @@ -21,7 +21,7 @@ use config::{ meta::stream::{FileMeta, StreamType}, utils::{ parquet::{VORTEX_FILE_META_KEY, encode_vortex_file_meta, new_parquet_writer}, - util::{DISTINCT_STREAM_PREFIX, is_trace_time_index_stream}, + util::is_trace_time_index_stream, }, }; use datafusion::{ @@ -108,20 +108,6 @@ pub async fn merge_parquet_files( format!( "SELECT MIN({TIMESTAMP_COL_NAME}) AS {TIMESTAMP_COL_NAME}, trace_id, MIN(min_ts) AS min_ts, MAX(max_ts) AS max_ts FROM tbl GROUP BY trace_id ORDER BY {TIMESTAMP_COL_NAME} DESC" ) - } else if cfg.limit.distinct_values_hourly - && stream_type == StreamType::Metadata - && stream_name.starts_with(DISTINCT_STREAM_PREFIX) - { - let fields = schema - .fields() - .iter() - .filter(|f| f.name() != TIMESTAMP_COL_NAME && f.name() != "count") - .map(|x| x.name().to_string()) - .collect::>(); - let fields_str = fields.join(", "); - format!( - "SELECT MIN({TIMESTAMP_COL_NAME}) AS {TIMESTAMP_COL_NAME}, SUM(count) as count, {fields_str} FROM tbl GROUP BY {fields_str} ORDER BY {TIMESTAMP_COL_NAME} DESC" - ) } else if stream_type == StreamType::Filelist { // for file list we do not have timestamp, so we instead sort by min ts of entries "SELECT * FROM tbl ORDER BY min_ts DESC".to_string() diff --git a/src/search/src/tantivy/search.rs b/src/search/src/tantivy/search.rs index 937f2e253d..3377cf17f8 100644 --- a/src/search/src/tantivy/search.rs +++ b/src/search/src/tantivy/search.rs @@ -308,6 +308,10 @@ impl TantivyResult { if ascend && distinct_values.len() >= limit { break; } + // null values are indexed as empty strings, skip them + if term.is_empty() { + continue; + } distinct_values.push(String::from_utf8(term.to_vec()).unwrap()); } } @@ -319,6 +323,10 @@ impl TantivyResult { if ascend && distinct_values.len() >= limit { break; } + // null values are indexed as empty strings, skip them + if term.is_empty() { + continue; + } distinct_values.push(String::from_utf8(term.to_vec()).unwrap()); } } @@ -395,6 +403,38 @@ mod tests { } } + #[test] + fn test_handle_simple_distinct_skips_null_terms() { + use crate::index::{Condition, IndexCondition}; + + let mut schema_builder = tantivy::schema::SchemaBuilder::new(); + let svc_field = schema_builder.add_text_field("svc", tantivy::schema::STRING); + let index = tantivy::index::Index::create_in_ram(schema_builder.build()); + let mut writer = index.writer_with_num_threads(1, 15_000_000).unwrap(); + // null values are indexed as empty strings (see tantivy_utils index_builder) + for svc in ["", "ziox", "", "ingress-nginx"] { + writer + .add_document(tantivy::doc!(svc_field => svc)) + .unwrap(); + } + writer.commit().unwrap(); + let searcher = index.reader().unwrap().searcher(); + + let mut condition = IndexCondition::new(); + condition.add_condition(Condition::All()); + let result = + TantivyResult::handle_simple_distinct(&searcher, &condition, "svc", 10, true).unwrap(); + match result { + TantivyResult::Distinct(values) => { + assert_eq!( + values, + HashSet::from(["ingress-nginx".to_string(), "ziox".to_string()]) + ); + } + other => panic!("expected Distinct, got {other:?}"), + } + } + #[test] fn test_handle_simple_select_missing_timestamp_returns_error() { let mut schema_builder = tantivy::schema::SchemaBuilder::new(); diff --git a/src/search_service/src/grpc/flight.rs b/src/search_service/src/grpc/flight.rs index 4122bb493f..43500c7916 100644 --- a/src/search_service/src/grpc/flight.rs +++ b/src/search_service/src/grpc/flight.rs @@ -215,10 +215,13 @@ pub async fn search( .as_ref() .map(|rule| rule.referenced_fields()) .unwrap_or_default(); + let distinct_fields_updated_at = + infra::db::tantivy_index::get_ttv_distinct_fields_updated_at().await; get_stream_setting_index_updated_at_for_fields( &stream_settings, stream_created_at, &index_used_fields, + distinct_fields_updated_at, ) }; diff --git a/tests/api-testing/tests/alerts/test_composite_alerts.py b/tests/api-testing/tests/alerts/test_composite_alerts.py index 7f6ec33e1d..817f1136cd 100644 --- a/tests/api-testing/tests/alerts/test_composite_alerts.py +++ b/tests/api-testing/tests/alerts/test_composite_alerts.py @@ -700,7 +700,7 @@ def test_child_rename_and_move_do_not_rewrite_stored_id_expression( client.delete(f"folders/alerts/{destination_folder}", prefix="api/v2/") -def test_openapi_exposes_composite_union_validate_preview_and_references( +def test_openapi_exposes_composite_create_validate_preview_and_references( client: OpenObserveClient, ): response = client.get("api-doc/openapi.json", prefix="") @@ -712,7 +712,11 @@ def test_openapi_exposes_composite_union_validate_preview_and_references( create_schema = paths["/api/v2/{org_id}/alerts"]["post"]["requestBody"][ "content" ]["application/json"]["schema"] - assert "oneOf" in create_schema or "anyOf" in create_schema + # the create request body documents the composite contract directly + # (alert_type discriminator + composite_condition), not a oneOf union + serialized_create = str(create_schema) + assert "alert_type" in serialized_create + assert "composite_condition" in serialized_create serialized = str(document["components"]["schemas"]) for token in ( "composite_condition",