mirror of
https://github.com/openobserve/openobserve.git
synced 2026-11-04 08:09:27 +00:00
fix: generate empty ttv file when index field not present in parquet (#9707)
# Fix: Create Empty Tantivy Index Files for Parquets Without Indexed Fields Scenarios: ### Context: - **Default FTS fields:** `log`, `message`, `msg`, `content`, `data`, `body`, `json`, `error`, `errors` (always included) - **Default secondary index fields:** `trace_id`, `service_name`, `operation_name` (always included) - **latest_schema:** Stream schema with all fields (including user-defined) - **Parquet data:** Actual data in the specific parquet file --- ### **Scenario 1:** All fields present in data - `latest_schema` has: default fields + user fields ✅ - FTS/Index settings: defaults + user configured fields ✅ - Parquet data has: ALL the configured fields ✅ - **Result:** Create .ttv with full data ✅ --- ### **Scenario 2:** Some configured fields missing from parquet data - `latest_schema` has: `body`, `continent`, `name` ✅ - FTS/Index settings: `body` (FTS), `continent`, `name` (index) ✅ - Parquet data has: `body`, `name` only (missing `continent`) ⚠️ - **Result:** Create .ttv with `body` and `name` data, `continent` as empty placeholder ✅ --- ### **Scenario 3:** All configured fields missing from parquet data (but parquet has other data) - `latest_schema` has: `body`, `continent` ✅ - FTS/Index settings: `body` (FTS), `continent` (index) ✅ - Parquet data has: `other_field_1`, `other_field_2` (10 rows, but no `body` or `continent`) ❌ - **Result:** Create .ttv as empty marker (has documents but no field data) ✅ --- ### **Scenario 4:** Parquet file is empty (0 rows) - `latest_schema` has: `body`, `continent` ✅ - FTS/Index settings: `body` (FTS), `continent` (index) ✅ - Parquet data has: 0 rows (empty file) ❌ - **Result:** Create .ttv as empty marker ✅ --- **Key Insight:** We only create empty .ttv files when the configured fields are valid in the stream schema (2). This ensures the .ttv file acts as a proper marker that "this parquet was checked for these fields, they just happen to be empty in this specific file." ## Problem When a stream has full-text search (FTS) or index fields configured in settings, but incoming data stops containing those fields, the system fails to create `.ttv` (tantivy index) files. This causes significant query performance degradation. ### Impact - Files **with** `.ttv` → fast tantivy index search - Files **without** `.ttv` → **slow DataFusion full table scan** with `LIKE '%pattern%'` ### Root Cause Two issues: 1. Code checked if configured fields exist in **parquet schema** (not stream schema) → skipped indexing if fields missing 2. Used parquet schema instead of stream schema for index generation → missing fields excluded from index **Result:** Parquets without indexed fields have `index_size = 0`, forcing expensive DataFusion scans during queries. ## Solution 1. **Always create `.ttv` files** when FTS/index fields are configured, even if parquet data doesn't contain those fields 2. **Use stream schema** (not parquet schema) for index generation to include all configured fields 3. Create empty index as marker: "this parquet was indexed but has no matching data" ## Test Scenarios ### Scenario 1: All Configured Fields Missing - **Config:** `body` for FTS - **Data:** No `body` field - **Result:** Empty `.ttv` created (0 documents) - **Query:** File skipped, no DataFusion scan ### Scenario 2: Partial Fields Missing (Secondary Index) - **Config:** `continent`, `name`, `flag_url` as secondary index - **Data:** Only `name` and `flag_url` present (no `continent`) - **Result:** `.ttv` created with all 3 fields: - `name`: indexed (has data) - `flag_url`: indexed (has data) - `continent`: empty column (no data) - **Query on `continent`:** Returns 0 matches (field exists but empty) ### **PR Type** Bug fix, Enhancement ___ ### **Description** - Align skip logic to stream‐configured index fields - Add `should_create_empty_index` flag in Tantivy module - Generate empty index file if no matching data fields - Enhance debug logs with empty‐index context ___ ### Diagram Walkthrough ```mermaid flowchart LR A["Stream settings"] --> B{"Configured FTS/index fields?"} B -- "No" --> C["Skip index generation"] B -- "Yes" --> D["merge_files calls generate_tantivy_index"] D --> E{"Schema contains fields?"} E -- "Yes" --> F["Build full index"] E -- "No" --> G["Build empty .ttv index"] ``` <details> <summary><h3> File Walkthrough</h3></summary> <table><thead><tr><th></th><th align="left">Relevant files</th></tr></thead><tbody><tr><td><strong>Enhancement</strong></td><td><table> <tr> <td> <details> <summary><strong>parquet.rs</strong><dd><code>Use configured fields check for indexing</code> </dd></summary> <hr> src/job/files/parquet.rs <ul><li>Removed schema‐field presence check<br> <li> Skip index only if no configured fields<br> <li> Updated debug log with config note</ul> </details> </td> <td><a href="https://github.com/openobserve/openobserve/pull/9707/files#diff-27f710981cfe1b938aeeb53abaed05961596c32e2edf3b8a2a59d3e72b2f542b">+6/-12</a> </td> </tr> <tr> <td> <details> <summary><strong>merge.rs</strong><dd><code>Simplify need_index evaluation</code> </dd></summary> <hr> src/service/compact/merge.rs <ul><li>Replace schema lookup with config‐based flag<br> <li> Skip logs now include config context<br> <li> Retain <code>need_index</code> logic for compactor</ul> </details> </td> <td><a href="https://github.com/openobserve/openobserve/pull/9707/files#diff-0ab016dbd5129c9c285c9c459bf78928147bce97dc92a03aa754da3da71168d7">+6/-10</a> </td> </tr> <tr> <td> <details> <summary><strong>mod.rs</strong><dd><code>Support empty index creation flag</code> </dd></summary> <hr> src/service/tantivy/mod.rs <ul><li>Introduce <code>should_create_empty_index</code> variable<br> <li> Only skip when no configured fields exist<br> <li> Log creation of empty index structure<br> <li> Update row‐count debug logging</ul> </details> </td> <td><a href="https://github.com/openobserve/openobserve/pull/9707/files#diff-68837d2f393b453002ec2d95f6882cfcaf70d9ef6547be2c9a890e8b0fc3c559">+17/-3</a> </td> </tr> </table></td></tr></tr></tbody></table> </details> ___
This commit is contained in:
Loakesh Indiran
@@ -967,13 +967,13 @@ async fn merge_files(
|
||||
}
|
||||
|
||||
// generate tantivy inverted index and write to storage
|
||||
let (schema, reader) = get_recordbatch_reader_from_bytes(&buf).await?;
|
||||
let (_parquet_schema, reader) = get_recordbatch_reader_from_bytes(&buf).await?;
|
||||
let index_size = create_tantivy_index(
|
||||
"INGESTER",
|
||||
&new_file_key,
|
||||
&full_text_search_fields,
|
||||
&index_fields,
|
||||
schema,
|
||||
latest_schema.clone(), // Use stream schema to include all configured fields
|
||||
reader,
|
||||
)
|
||||
.await
|
||||
|
||||
@@ -823,6 +823,9 @@ pub async fn merge_files(
|
||||
.values()
|
||||
.flat_map(|s| s.fields().iter().map(|f| f.name().to_string()))
|
||||
.collect::<HashSet<_>>();
|
||||
// Keep original schema for index generation (has all stream fields)
|
||||
let stream_schema_for_index = Arc::new(latest_schema.clone());
|
||||
// Create filtered schema for parquet merging (only fields in parquet files)
|
||||
let latest_schema = Arc::new(latest_schema.retain(all_fields));
|
||||
let mut latest_schema_fields = HashMap::with_capacity(latest_schema.fields().len());
|
||||
for field in latest_schema.fields() {
|
||||
@@ -980,6 +983,7 @@ pub async fn merge_files(
|
||||
&index_fields,
|
||||
&retain_file_list,
|
||||
&mut new_file_meta,
|
||||
stream_schema_for_index.clone(),
|
||||
&buf,
|
||||
)
|
||||
.await?;
|
||||
@@ -1020,6 +1024,7 @@ pub async fn merge_files(
|
||||
&index_fields,
|
||||
&retain_file_list,
|
||||
&mut new_file_meta,
|
||||
stream_schema_for_index.clone(),
|
||||
&buf,
|
||||
)
|
||||
.await?;
|
||||
@@ -1050,15 +1055,16 @@ async fn generate_inverted_index(
|
||||
index_fields: &[String],
|
||||
retain_file_list: &[FileKey],
|
||||
new_file_meta: &mut FileMeta,
|
||||
latest_schema: Arc<Schema>,
|
||||
buf: &Bytes,
|
||||
) -> Result<(), anyhow::Error> {
|
||||
let (schema, reader) = get_recordbatch_reader_from_bytes(buf).await?;
|
||||
let (_parquet_schema, reader) = get_recordbatch_reader_from_bytes(buf).await?;
|
||||
let index_size = create_tantivy_index(
|
||||
"COMPACTOR",
|
||||
new_file_key,
|
||||
full_text_search_fields,
|
||||
index_fields,
|
||||
schema,
|
||||
latest_schema, // Use stream schema to include all configured fields
|
||||
reader,
|
||||
)
|
||||
.await
|
||||
|
||||
+102
-8
@@ -132,7 +132,7 @@ pub(crate) async fn generate_tantivy_index<D: tantivy::Directory>(
|
||||
.union(&index_fields)
|
||||
.cloned()
|
||||
.collect::<HashSet<_>>();
|
||||
// no fields need to create index, return
|
||||
|
||||
if tantivy_fields.is_empty() {
|
||||
return Ok(None);
|
||||
}
|
||||
@@ -300,11 +300,13 @@ pub(crate) async fn generate_tantivy_index<D: tantivy::Directory>(
|
||||
}
|
||||
}
|
||||
let total_num_rows = task.await??;
|
||||
// no docs need to create index, return
|
||||
if total_num_rows == 0 {
|
||||
return Ok(None);
|
||||
}
|
||||
log::debug!("write documents to tantivy index success");
|
||||
// Create index even with 0 rows since we have valid configured fields in stream schema
|
||||
// (empty index acts as a marker to prevent expensive DataFusion scans)
|
||||
log::debug!(
|
||||
"write documents to tantivy index success (rows: {}, empty_index: {})",
|
||||
total_num_rows,
|
||||
total_num_rows == 0
|
||||
);
|
||||
|
||||
let index = tokio::task::spawn_blocking(move || {
|
||||
index_writer.finalize().map_err(|e| {
|
||||
@@ -436,7 +438,8 @@ mod tests {
|
||||
.await;
|
||||
|
||||
assert!(result.is_ok());
|
||||
assert!(result.unwrap().is_none()); // Should return None for empty data
|
||||
// Should create empty index when fields are configured (even with 0 rows)
|
||||
assert!(result.unwrap().is_some());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -532,7 +535,7 @@ mod tests {
|
||||
let batch = create_test_batch(10, true, true, true);
|
||||
let stream = create_test_stream(vec![batch.clone()]).await;
|
||||
|
||||
// Request fields that don't exist in the schema
|
||||
// Request fields that don't exist in the schema at all
|
||||
let result = generate_tantivy_index(
|
||||
dir,
|
||||
stream,
|
||||
@@ -544,6 +547,8 @@ mod tests {
|
||||
|
||||
assert!(result.is_ok());
|
||||
let index = result.unwrap();
|
||||
// Should NOT create index when configured fields don't exist in stream schema
|
||||
// (this indicates a configuration error, not just missing data)
|
||||
assert!(index.is_none());
|
||||
}
|
||||
|
||||
@@ -773,6 +778,95 @@ mod tests {
|
||||
assert_eq!(result.unwrap(), 0); // Should return 0 when filename conversion fails
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_generate_tantivy_index_stream_schema_vs_parquet_schema() {
|
||||
// This test validates the critical fix where we pass stream schema (with all
|
||||
// configured fields) instead of parquet schema (only fields in actual data).
|
||||
//
|
||||
// Scenario: Stream settings configure `continent`, `name`, `flag_url` as secondary
|
||||
// index fields, but the parquet data only contains `name` and `flag_url`.
|
||||
// The missing field `continent` should still be included in the .ttv schema.
|
||||
|
||||
let dir = RamDirectory::create();
|
||||
|
||||
// Create parquet data with only `name` and `flag_url` (no `continent`)
|
||||
let parquet_fields = vec![
|
||||
Field::new(TIMESTAMP_COL_NAME, DataType::Int64, false),
|
||||
Field::new("name", DataType::Utf8, false),
|
||||
Field::new("flag_url", DataType::Utf8, false),
|
||||
];
|
||||
let parquet_schema = Arc::new(Schema::new(parquet_fields));
|
||||
|
||||
let timestamps: Vec<i64> = (0..10).map(|i| 1000 + i as i64).collect();
|
||||
let names: Vec<String> = (0..10).map(|i| format!("Name {i}")).collect();
|
||||
let flag_urls: Vec<String> = (0..10)
|
||||
.map(|i| format!("https://example.com/{i}"))
|
||||
.collect();
|
||||
|
||||
let parquet_batch = RecordBatch::try_new(
|
||||
parquet_schema.clone(),
|
||||
vec![
|
||||
Arc::new(Int64Array::from(timestamps)),
|
||||
Arc::new(StringArray::from(names)),
|
||||
Arc::new(StringArray::from(flag_urls)),
|
||||
],
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
// Create stream schema with all configured fields including `continent`
|
||||
let stream_fields = vec![
|
||||
Field::new(TIMESTAMP_COL_NAME, DataType::Int64, false),
|
||||
Field::new("continent", DataType::Utf8, false), // Configured but not in data
|
||||
Field::new("name", DataType::Utf8, false),
|
||||
Field::new("flag_url", DataType::Utf8, false),
|
||||
];
|
||||
let stream_schema = Arc::new(Schema::new(stream_fields));
|
||||
|
||||
let stream = create_test_stream(vec![parquet_batch]).await;
|
||||
|
||||
// Generate index with stream schema (all configured fields)
|
||||
let result = generate_tantivy_index(
|
||||
dir,
|
||||
stream,
|
||||
&[], // No FTS fields
|
||||
&[
|
||||
"continent".to_string(),
|
||||
"name".to_string(),
|
||||
"flag_url".to_string(),
|
||||
],
|
||||
stream_schema.clone(), // Pass stream schema, not parquet schema
|
||||
)
|
||||
.await;
|
||||
|
||||
assert!(result.is_ok());
|
||||
let index = result.unwrap();
|
||||
assert!(index.is_some());
|
||||
|
||||
let index = index.unwrap();
|
||||
let tantivy_schema = index.schema();
|
||||
|
||||
// Verify that ALL configured fields are in the tantivy schema,
|
||||
// including `continent` which is missing from the parquet data
|
||||
assert!(
|
||||
tantivy_schema.get_field("continent").is_ok(),
|
||||
"continent field should be in tantivy schema even though it's not in parquet data"
|
||||
);
|
||||
assert!(
|
||||
tantivy_schema.get_field("name").is_ok(),
|
||||
"name field should be in tantivy schema"
|
||||
);
|
||||
assert!(
|
||||
tantivy_schema.get_field("flag_url").is_ok(),
|
||||
"flag_url field should be in tantivy schema"
|
||||
);
|
||||
assert!(tantivy_schema.get_field(TIMESTAMP_COL_NAME).is_ok());
|
||||
|
||||
// Verify we have 10 documents
|
||||
let reader = index.reader().unwrap();
|
||||
let searcher = reader.searcher();
|
||||
assert_eq!(searcher.num_docs(), 10);
|
||||
}
|
||||
|
||||
// Note: Full testing of create_tantivy_index with storage operations would require
|
||||
// mocking the storage layer, which is complex for unit tests. The main logic is
|
||||
// tested in generate_tantivy_index. Integration tests would be more appropriate
|
||||
|
||||
Reference in New Issue
Block a user