From 781f81d4e428d8a4ffeb02b220a1f6e50db25ba3 Mon Sep 17 00:00:00 2001 From: Loakesh Indiran <66478092+Loaki07@users.noreply.github.com> Date: Wed, 24 Dec 2025 12:44:00 +0530 Subject: [PATCH] fix: generate empty ttv file when index field not present in parquet (#9707) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit # 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 :white_check_mark: - FTS/Index settings: defaults + user configured fields :white_check_mark: - Parquet data has: ALL the configured fields :white_check_mark: - **Result:** Create .ttv with full data :white_check_mark: --- ### **Scenario 2:** Some configured fields missing from parquet data - `latest_schema` has: `body`, `continent`, `name` :white_check_mark: - FTS/Index settings: `body` (FTS), `continent`, `name` (index) :white_check_mark: - Parquet data has: `body`, `name` only (missing `continent`) :warning: - **Result:** Create .ttv with `body` and `name` data, `continent` as empty placeholder :white_check_mark: --- ### **Scenario 3:** All configured fields missing from parquet data (but parquet has other data) - `latest_schema` has: `body`, `continent` :white_check_mark: - FTS/Index settings: `body` (FTS), `continent` (index) :white_check_mark: - Parquet data has: `other_field_1`, `other_field_2` (10 rows, but no `body` or `continent`) :x: - **Result:** Create .ttv as empty marker (has documents but no field data) :white_check_mark: --- ### **Scenario 4:** Parquet file is empty (0 rows) - `latest_schema` has: `body`, `continent` :white_check_mark: - FTS/Index settings: `body` (FTS), `continent` (index) :white_check_mark: - Parquet data has: 0 rows (empty file) :x: - **Result:** Create .ttv as empty marker :white_check_mark: --- **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"] ```

File Walkthrough
Relevant files
Enhancement
parquet.rs
Use configured fields check for indexing                                 
src/job/files/parquet.rs
  • Removed schema‐field presence check
  • Skip index only if no configured fields
  • Updated debug log with config note
 +6/-12   
merge.rs
Simplify need_index evaluation                                                     
src/service/compact/merge.rs
  • Replace schema lookup with config‐based flag
  • Skip logs now include config context
  • Retain need_index logic for compactor
 +6/-10   
mod.rs
Support empty index creation flag                                               
src/service/tantivy/mod.rs
  • Introduce should_create_empty_index variable
  • Only skip when no configured fields exist
  • Log creation of empty index structure
  • Update row‐count debug logging
 +17/-3   
___ --- src/job/files/parquet.rs | 4 +- src/service/compact/merge.rs | 10 +++- src/service/tantivy/mod.rs | 110 ++++++++++++++++++++++++++++++++--- 3 files changed, 112 insertions(+), 12 deletions(-) diff --git a/src/job/files/parquet.rs b/src/job/files/parquet.rs index 1727ea3669..59fe927760 100644 --- a/src/job/files/parquet.rs +++ b/src/job/files/parquet.rs @@ -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 diff --git a/src/service/compact/merge.rs b/src/service/compact/merge.rs index aafbf163e3..5b43c10fe5 100644 --- a/src/service/compact/merge.rs +++ b/src/service/compact/merge.rs @@ -823,6 +823,9 @@ pub async fn merge_files( .values() .flat_map(|s| s.fields().iter().map(|f| f.name().to_string())) .collect::>(); + // 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, 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 diff --git a/src/service/tantivy/mod.rs b/src/service/tantivy/mod.rs index d260f68cb6..c154f7be82 100644 --- a/src/service/tantivy/mod.rs +++ b/src/service/tantivy/mod.rs @@ -132,7 +132,7 @@ pub(crate) async fn generate_tantivy_index( .union(&index_fields) .cloned() .collect::>(); - // 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( } } 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 = (0..10).map(|i| 1000 + i as i64).collect(); + let names: Vec = (0..10).map(|i| format!("Name {i}")).collect(); + let flag_urls: Vec = (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