starred/openobserve
1
0
Fork 0
mirror of https://github.com/openobserve/openobserve.git synced 2026-11-04 08:09:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: replace distinct_values_* streams with tantivy TopK for the _values API (#13834)

Browse Source 
Design at: #13822

## Summary

`distinct_values_*` metadata streams existed to serve one reader: the
`_values` API. They are a redundant — and lossier — materialization of
what the tantivy secondary index already holds, and the index has the
collectors needed to replace them outright (`SimpleTopN` /
`SimpleDistinct`). This PR removes the distinct streams from the write
and read paths and lets the `_values` API be served by the tantivy index
optimizer.

### 1. Treat distinct value fields as secondary index fields (read-time
union)

- `get_stream_setting_index_fields()` — the single accessor used by both
the index build side (ingester parquet/pack jobs, compactor merge) and
the query side (leader/follower index optimizers) — now unions
`distinct_value_fields` into the effective index field set, gated on
`enable_distinct_fields`. Fields that cannot be secondary-indexed (full
text search keys, partition keys, reserved columns) are skipped; the
values API falls back to a normal scan for those.
- The distinct list is deliberately **not merged into
`settings.index_fields`**: the two stay independent, so removing a
distinct value field (or disabling `enable_distinct_fields`) cleanly
stops its indexing with no residue, and the user-managed `index_fields`
list is never mutated behind their back.
- Correctness gate: a distinct-only field is considered indexed since
its `added_ts` (`get_stream_setting_index_updated_at_for_fields`), so
parquet files older than that are served by DataFusion scan instead of
the index (`split_file_list_by_time_range`). `added_ts` is set on add
and reset when `enable_distinct_fields` is re-enabled (existing
behavior). For streams configured before this release the fields only
start being indexed at upgrade time, so `added_ts` is additionally
clamped by a release-wide cutoff `/tantivy/distinct_fields/updated_at` —
a lazy marker initialized to the first access after upgrade, following
the existing `ttv_secondary_index_updated_at` pattern. No data migration
and no `DB_SCHEMA_VERSION` bump needed.

### 2. `_values` API always queries the original stream

- Removed `can_use_distinct_stream()` and the `distinct_values_*` /
`SUM(count)` / `StreamType::Metadata` rewrite from `values_inner` (GET
`/_values`, formerly `values_v1`) and `build_search_request_per_field`
(POST `/_values_stream`).
- The generated SQL shapes are exactly what the index optimizer detects:
- `SELECT f AS zo_sql_key, COUNT(*) AS zo_sql_num FROM s GROUP BY
zo_sql_key ORDER BY zo_sql_num DESC, zo_sql_key ASC` + limit →
`SimpleTopN`
- `no_count`: `SELECT f AS zo_sql_key FROM s GROUP BY zo_sql_key ORDER
BY zo_sql_key ASC` + limit → `SimpleDistinct`
- The `keyword` filter (`str_match_ignore_case`) is a valid index
condition, so keyword-filtered values queries stay on the index path
too.

### 3. Remove distinct values generation

- Dropped the per-record distinct value collection in logs and traces
ingestion and the whole `core::metadata` module (channel, mem-table,
flush job, shutdown hook).
- Removed the now-unused `DISTINCT_FIELDS` static and the
`ZO_FEATURE_DISTINCT_EXTRA_FIELDS` / `ZO_DISTINCT_VALUES_INTERVAL` /
`ZO_DISTINCT_VALUES_HOURLY` config knobs, including the `SUM(count)`
merge rollup the last one gated — legacy `distinct_values_*` streams now
compact as ordinary metadata streams. The default fields
`DISTINCT_FIELDS` carried (`service_name`, `operation_name`) are already
default secondary index fields via `SQL_SECONDARY_INDEX_SEARCH_FIELDS`,
so traces `_values` queries on them remain index-served.
- Renamed `values_v1` to `values_inner`: the `v2` variant was removed
long ago (#5087) and the distinct stream switch inside `v1` is gone now
too, so the version suffix was pure history.

### 4. Default fields after this change

- `service_name`, `operation_name`: default secondary index fields
(unchanged, via `SQL_SECONDARY_INDEX_SEARCH_FIELDS`).
- `trace_id`: now also a built-in default **bloom filter** field
(`BLOOM_FILTER_DEFAULT_FIELDS`, gated on
`ZO_FEATURE_DEFAULT_INDEX_FIELDS_ENABLED`), in addition to being a
default index field — bloom sits on top of the secondary index and
prunes files for high-cardinality equality lookups.
- There is no "default distinct fields" concept anymore.

### 5. Fix surfaced by e2e: SimpleDistinct returned null values as `""`

The tantivy index encodes null/missing field values as empty strings
(`index_builder` adds `""` for null docs), so the `SimpleDistinct`
term-dictionary walk returned `""` as a distinct value — sorted first in
ascending order, it surfaced as a blank first option in dashboard
variable dropdowns, which the DataFusion `GROUP BY` path never returns
(its null group sorts last). This was exposed once distinct fields
became index-served; `handle_simple_distinct` now skips empty terms
(with a unit test).

### 6. Drive-by: fix the failing `api_fuzz_tests` CI job

`CreateAlertRequestSchema` was an untagged documentation union with two
identical variants, producing `oneOf: [CreateAlertRequestBody,
CreateAlertRequestBody]` in the OpenAPI spec. Under `oneOf` semantics
(exactly one branch must match) no instance can ever validate, so
schemathesis filtered out every generated example and `api_fuzz_tests`
failed on `POST /api/v2/{org_id}/alerts` on every CI run since that
schema landed. The handler deserializes `CreateAlertRequestBody`
directly, so the annotation now documents exactly that type; runtime
behavior is unchanged.

## What stays (intentionally)

- `distinct_value_fields` / `enable_distinct_fields` in stream settings:
still the user-facing way to declare values-API fields; the list now
drives the read-time index-field union described above.
- The `distinct_value_fields` DB table that tracks dashboard/report
usage of these fields (still gates removal in the settings API).
- `DISTINCT_STREAM_PREFIX` helpers, retention sync and stream-deletion
cleanup for existing `distinct_values_*` streams — legacy stream data
keeps compacting (as ordinary metadata streams) and ages out via
retention; no new records are written to them.

## Testing

- `cargo check --workspace --all-targets` and `cargo clippy --workspace
--all-targets` clean.
- Enterprise build verified with `Cargo.toml.openobserve` (`cargo check
--all-targets`).
- New unit tests for the index-field union (skip rules for
FTS/partition/reserved fields, `enable_distinct_fields` gating) and the
`added_ts` / release-cutoff indexed-since resolution; schema, infra,
config and api-search unit tests pass. A few pre-existing failures on a
clean tree (config `ssrf_guard`, one flaky sqlite-backed destinations
test, one `.env`-dependent index-fields test) are local-environment
issues and unrelated.

Closes #13822
This commit is contained in:
Hengfei Yang
2026-08-16 15:19:31 +00:00
committed by GitHub
parent 7f74925605
commit 41ff3cd9b6
16 changed files with 270 additions and 870 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api/management/src/models/alerts/requests.rs
-10
View File
@@ -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 {
src/api/management/src/request/alerts/mod.rs
+3 -3
View File
@@ -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<String>, 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 = ()),
src/api/search/src/search/mod.rs
+21 -188
View File
@@ -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<String> =
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::<Vec<_>>();
if columns.len() < 2 {
return Err(Error::other("Invalid filter format"));
}
let vals = columns[1].split(',').collect::<Vec<_>>().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(),
src/config/src/config.rs
+5 -30
View File
@@ -217,18 +217,18 @@ pub static QUICK_MODEL_FIELDS: Lazy<Vec<String>> = Lazy::new(|| {
fields
});
const _DEFAULT_DISTINCT_FIELDS: [&str; 2] = ["service_name", "operation_name"];
pub static DISTINCT_FIELDS: Lazy<Vec<String>> = Lazy::new(|| {
const _DEFAULT_BLOOM_FILTER_FIELDS: [&str; 1] = ["trace_id"];
pub static BLOOM_FILTER_DEFAULT_FIELDS: Lazy<Vec<String>> = 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<Vec<String>> = Lazy::new(|| {
fields
});
pub static BLOOM_FILTER_DEFAULT_FIELDS: Lazy<Vec<String>> = 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::<Vec<_>>();
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(
src/core/src/lib.rs
-1
View File
@@ -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;
src/core/src/logs/mod.rs
+1 -38
View File
@@ -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<String, SchemaRecords> = 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 });
src/core/src/metadata/distinct_values.rs
-417
View File
@@ -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 <http://www.gnu.org/licenses/>.
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<DistinctValues> = Lazy::new(DistinctValues::new);
type MemTable = FxIndexMap<String, FxIndexMap<DvItem, u32>>;
pub struct DistinctValues {
channel: Arc<mpsc::Sender<DvEvent>>,
shutdown: Arc<AtomicBool>,
mem_table: Arc<RwLock<MemTable>>,
}
#[derive(Debug, Default, Eq, Hash, PartialEq, Clone, Serialize, Deserialize)]
pub struct DvItem {
pub stream_type: StreamType,
pub stream_name: String,
pub value: Map<String, Value>,
}
#[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<mpsc::Sender<DvEvent>> {
let (tx, mut rx) = mpsc::channel::<DvEvent>(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<Schema> {
// 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<MetadataItem>) -> Result<()> {
let mut group_items: FxIndexMap<DvItem, u32> = 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<String, Value>, 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<String, SchemaRecords> = 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());
}
}
src/core/src/metadata/mod.rs
-106
View File
@@ -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 <http://www.gnu.org/licenses/>.
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<MetadataManager> = 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<Schema>;
fn write(
&self,
org_id: &str,
data: Vec<MetadataItem>,
) -> impl std::future::Future<Output = infra::errors::Result<()>> + Send;
fn flush(&self) -> impl std::future::Future<Output = infra::errors::Result<()>> + Send;
fn stop(&self) -> impl std::future::Future<Output = infra::errors::Result<()>> + 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<MetadataItem>,
) -> 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));
}
}
src/core/src/traces/mod.rs
+2 -39
View File
@@ -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<StreamPartition> = 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<String, SchemaRecords> = 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;
src/infra/src/db/tantivy_index.rs
+28 -6
View File
@@ -17,30 +17,52 @@ use tokio::sync::OnceCell;
static TIMESTAMP_UPDATED_AT: OnceCell<i64> = OnceCell::const_new();
static SECONDARY_INDEX_UPDATED_AT: OnceCell<i64> = OnceCell::const_new();
static DISTINCT_FIELDS_UPDATED_AT: OnceCell<i64> = 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::<i64>().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
}
}
}
src/infra/src/schema/mod.rs
+159 -12
View File
@@ -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<T: std::borrow::Borrow<StreamSettings>>(
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<T: std::borrow::Borrow<Str
settings: &Option<T>,
created_at: Option<i64>,
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<T: std::borrow::Borrow<Str
.map(|f| {
if SQL_SECONDARY_INDEX_SEARCH_FIELDS.contains(f) {
BASE_TIME.timestamp_micros()
} else if let Some(updated_at) = settings
.as_ref()
.and_then(|s| s.borrow().index_fields_updated_at.get(f))
.copied()
{
updated_at
} else if let Some(added_ts) = settings.as_ref().and_then(|s| {
let s = s.borrow();
if s.enable_distinct_fields && !s.index_fields.contains(f) {
s.distinct_value_fields
.iter()
.find(|field| field.name == *f)
.map(|field| field.added_ts.max(distinct_fields_updated_at))
} else {
None
}
}) {
added_ts
} else {
settings
.as_ref()
.and_then(|s| s.borrow().index_fields_updated_at.get(f))
.copied()
.unwrap_or(default_updated_at)
default_updated_at
}
})
.max()
@@ -1230,6 +1270,111 @@ mod tests {
assert_eq!(unique_count, fields.len());
}
#[test]
fn test_get_stream_setting_index_fields_includes_distinct_fields() {
use config::meta::stream::{DistinctField, StreamPartition};
let make_distinct = |name: &str| DistinctField {
name: name.to_string(),
added_ts: 1,
};
let settings = StreamSettings {
index_fields: vec!["index_field".to_string()],
full_text_search_keys: vec!["fts".to_string()],
partition_keys: vec![StreamPartition::new("pk")],
distinct_value_fields: vec![
make_distinct("svc"),
// full text search keys cannot be secondary index fields
make_distinct("fts"),
// partition keys cannot be secondary index fields
make_distinct("pk"),
// reserved columns cannot be secondary index fields
make_distinct(TIMESTAMP_COL_NAME),
],
..Default::default()
};
let fields = get_stream_setting_index_fields(&Some(settings.clone()));
assert!(fields.contains(&"index_field".to_string()));
assert!(fields.contains(&"svc".to_string()));
assert!(!fields.contains(&"fts".to_string()));
assert!(!fields.contains(&"pk".to_string()));
assert!(!fields.contains(&TIMESTAMP_COL_NAME.to_string()));
let unique_count = fields.iter().collect::<hashbrown::HashSet<_>>().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::<StreamSettings>,
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::<StreamSettings>,
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);
}
src/main.rs
+1 -3
View File
@@ -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
src/search/src/datafusion/merge/mod.rs
+1 -15
View File
@@ -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::<Vec<_>>();
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()
src/search/src/tantivy/search.rs
+40
View File
@@ -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();
src/search_service/src/grpc/flight.rs
+3
View File
@@ -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,
)
};
tests/api-testing/tests/alerts/test_composite_alerts.py
+6 -2
View File
@@ -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",