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