mirror of
https://github.com/openobserve/openobserve.git
synced 2026-11-03 16:10:21 +00:00
Generated
+1
@@ -11074,6 +11074,7 @@ dependencies = [
|
||||
"rayon",
|
||||
"regex",
|
||||
"regex-syntax",
|
||||
"schema",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"sqlparser",
|
||||
|
||||
@@ -208,8 +208,19 @@ pub static SQL_SECONDARY_INDEX_SEARCH_FIELDS: Lazy<Vec<String>> = Lazy::new(|| {
|
||||
fields
|
||||
});
|
||||
|
||||
// Losing these silently degrades sourcemap translation, breadcrumbs and session replay.
|
||||
const _DEFAULT_QUICK_MODE_FIELDS: [&str; 4] = ["service", "version", "session_id", "view_url"];
|
||||
const _DEFAULT_QUICK_MODE_FIELDS: [&str; 9] = [
|
||||
// Losing these silently degrades sourcemap translation, breadcrumbs and session replay.
|
||||
"service",
|
||||
"version",
|
||||
"session_id",
|
||||
"view_url",
|
||||
// Losing these leaves the trace detail page without spans to build a waterfall from.
|
||||
"service_name",
|
||||
"operation_name",
|
||||
"trace_id",
|
||||
"span_id",
|
||||
"duration",
|
||||
];
|
||||
pub static QUICK_MODEL_FIELDS: Lazy<Vec<String>> = Lazy::new(|| {
|
||||
let mut fields = chain(
|
||||
_DEFAULT_QUICK_MODE_FIELDS.iter().map(|s| s.to_string()),
|
||||
|
||||
@@ -50,6 +50,7 @@ rand.workspace = true
|
||||
rayon.workspace = true
|
||||
regex.workspace = true
|
||||
regex-syntax.workspace = true
|
||||
schema.workspace = true
|
||||
serde.workspace = true
|
||||
serde_json.workspace = true
|
||||
sqlparser.workspace = true
|
||||
|
||||
@@ -219,7 +219,7 @@ impl Sql {
|
||||
&columns,
|
||||
has_original_column,
|
||||
query.quick_mode || cfg.limit.quick_mode_force_enabled,
|
||||
cfg.limit.quick_mode_num_fields,
|
||||
stream_type,
|
||||
&search_event_type,
|
||||
match_visitor.has_match_all,
|
||||
);
|
||||
|
||||
@@ -18,7 +18,7 @@ use std::sync::Arc;
|
||||
use arrow_schema::FieldRef;
|
||||
use config::{
|
||||
ALL_VALUES_COL_NAME, ID_COL_NAME, ORIGINAL_DATA_COL_NAME, TIMESTAMP_COL_NAME, get_config,
|
||||
meta::search::SearchEventType,
|
||||
meta::{search::SearchEventType, sql::TableReferenceExt, stream::StreamType},
|
||||
};
|
||||
use datafusion::{arrow::datatypes::Schema, common::TableReference};
|
||||
use hashbrown::{HashMap, HashSet};
|
||||
@@ -26,17 +26,19 @@ use infra::schema::{
|
||||
SchemaCache, get_stream_setting_defined_schema_fields, get_stream_setting_fts_fields,
|
||||
unwrap_stream_settings,
|
||||
};
|
||||
use schema::check_schema_for_defined_schema_fields;
|
||||
|
||||
pub fn generate_select_star_schema(
|
||||
schemas: HashMap<TableReference, Arc<SchemaCache>>,
|
||||
columns: &HashMap<TableReference, HashSet<String>>,
|
||||
has_original_column: HashMap<TableReference, bool>,
|
||||
quick_mode: bool,
|
||||
quick_mode_num_fields: usize,
|
||||
stream_type: StreamType,
|
||||
search_event_type: &Option<SearchEventType>,
|
||||
need_fst_fields: bool,
|
||||
) -> HashMap<TableReference, Arc<SchemaCache>> {
|
||||
let cfg = get_config();
|
||||
let quick_mode_num_fields = cfg.limit.quick_mode_num_fields;
|
||||
let mut used_schemas = HashMap::new();
|
||||
for (name, schema) in schemas {
|
||||
let stream_settings = unwrap_stream_settings(schema.schema());
|
||||
@@ -71,6 +73,7 @@ pub fn generate_select_star_schema(
|
||||
&fts_fields,
|
||||
skip_original_column,
|
||||
need_fst_fields,
|
||||
name.get_stream_type(stream_type),
|
||||
)
|
||||
} else {
|
||||
// skip selecting "_original" column if `SELECT * ...`
|
||||
@@ -133,6 +136,7 @@ pub fn generate_quick_mode_fields(
|
||||
fts_fields: &[String],
|
||||
skip_original_column: bool,
|
||||
need_fst_fields: bool,
|
||||
stream_type: StreamType,
|
||||
) -> Vec<Arc<arrow_schema::Field>> {
|
||||
let cfg = get_config();
|
||||
let strategy = cfg.limit.quick_mode_strategy.to_lowercase();
|
||||
@@ -224,6 +228,16 @@ pub fn generate_quick_mode_fields(
|
||||
}
|
||||
}
|
||||
|
||||
// truncation must not drop the columns a stream type is rendered from
|
||||
for field in check_schema_for_defined_schema_fields(stream_type, schema, Vec::new()) {
|
||||
if !fields_name.contains(&field)
|
||||
&& let Ok(field) = schema.field_with_name(&field)
|
||||
{
|
||||
fields.push(Arc::new(field.clone()));
|
||||
fields_name.insert(field.name().to_string());
|
||||
}
|
||||
}
|
||||
|
||||
// include gen AI fields for LLM streams
|
||||
if let Some(settings) = unwrap_stream_settings(schema)
|
||||
&& settings.is_llm_stream
|
||||
@@ -419,7 +433,7 @@ mod tests {
|
||||
let schema = Schema::new(fields);
|
||||
|
||||
// Mock config - default strategy is "first"
|
||||
let result = generate_quick_mode_fields(&schema, None, &[], false, false);
|
||||
let result = generate_quick_mode_fields(&schema, None, &[], false, false, StreamType::Logs);
|
||||
|
||||
// Should include timestamp and some fields (exact count depends on config)
|
||||
assert!(!result.is_empty());
|
||||
@@ -439,7 +453,8 @@ mod tests {
|
||||
let mut columns = HashSet::new();
|
||||
columns.insert("selected_field".to_string());
|
||||
|
||||
let result = generate_quick_mode_fields(&schema, Some(columns), &[], false, false);
|
||||
let result =
|
||||
generate_quick_mode_fields(&schema, Some(columns), &[], false, false, StreamType::Logs);
|
||||
|
||||
// Should include timestamp and selected field
|
||||
assert!(result.iter().any(|f| f.name() == TIMESTAMP_COL_NAME));
|
||||
@@ -456,11 +471,40 @@ mod tests {
|
||||
fields.push(Arc::new(Field::new("service", DataType::Utf8, true)));
|
||||
let schema = Schema::new(fields);
|
||||
|
||||
let result = generate_quick_mode_fields(&schema, None, &[], false, false);
|
||||
let result = generate_quick_mode_fields(&schema, None, &[], false, false, StreamType::Logs);
|
||||
|
||||
assert!(result.iter().any(|f| f.name() == "service"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_generate_quick_mode_fields_keeps_trace_fields() {
|
||||
// Not in the built-in default list, so only the stream type guard can bring them back.
|
||||
let trace_fields = [
|
||||
"span_status",
|
||||
"span_kind",
|
||||
"events",
|
||||
"reference_parent_span_id",
|
||||
];
|
||||
let over_cutoff = get_config().limit.quick_mode_num_fields + 100;
|
||||
let mut fields = (0..over_cutoff)
|
||||
.map(|i| Arc::new(Field::new(format!("field{i}"), DataType::Utf8, true)))
|
||||
.collect::<Vec<_>>();
|
||||
for name in trace_fields {
|
||||
fields.push(Arc::new(Field::new(name, DataType::Utf8, true)));
|
||||
}
|
||||
let schema = Schema::new(fields);
|
||||
|
||||
let result =
|
||||
generate_quick_mode_fields(&schema, None, &[], false, false, StreamType::Traces);
|
||||
|
||||
for name in trace_fields {
|
||||
assert!(result.iter().any(|f| f.name() == name), "missing {name}");
|
||||
}
|
||||
|
||||
let logs = generate_quick_mode_fields(&schema, None, &[], false, false, StreamType::Logs);
|
||||
assert!(!logs.iter().any(|f| f.name() == "span_status"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_generate_quick_mode_fields_skip_original() {
|
||||
let fields = vec![
|
||||
@@ -476,6 +520,7 @@ mod tests {
|
||||
&[],
|
||||
true, // skip_original_column = true
|
||||
false,
|
||||
StreamType::Logs,
|
||||
);
|
||||
|
||||
// Should not include original data column
|
||||
@@ -501,6 +546,7 @@ mod tests {
|
||||
&fts_fields,
|
||||
false,
|
||||
true, // need_fst_fields = true
|
||||
StreamType::Logs,
|
||||
);
|
||||
|
||||
// Should include FTS fields
|
||||
|
||||
Reference in New Issue
Block a user