fix: inverted index create error when the recordbatch is empty (#3544)

src/config/src/utils/arrow.rs

@@ -22,6 +22,9 @@ use super::json::{Map as JsonMap, Value};
 pub fn record_batches_to_json_rows(
     batches: &[&RecordBatch],
 ) -> Result<Vec<JsonMap<String, Value>>, anyhow::Error> {
+    if batches.is_empty() || batches.iter().all(|b| b.num_rows() == 0) {
+        return Ok(Vec::new());
+    }
     let json_buf = Vec::with_capacity(
         batches
             .iter()

src/infra/src/schema/mod.rs

@@ -389,20 +389,6 @@ pub async fn merge(
         }),
     )
     .await?;
-
-    // super cluster
-    #[cfg(feature = "enterprise")]
-    if O2_CONFIG.super_cluster.enabled {
-        o2_enterprise::enterprise::super_cluster::queue::schema_merge(
-            &key_for_update,
-            json::to_vec(&schema).unwrap().into(),
-            infra::db::NEED_WATCH,
-            start_dt,
-        )
-        .await
-        .map_err(|e| Error::Message(e.to_string()))?;
-    }
-
     rx.await.map_err(|e| Error::Message(e.to_string()))
 }
 

src/job/files/parquet.rs

@@ -624,8 +624,8 @@ pub(crate) async fn generate_index_on_ingester(
 
     let index_record_batches =
         prepare_index_record_batches_v1(buf, org_id, stream_name, &new_file_key).await?;
-    let record_batches: Vec<&RecordBatch> = index_record_batches.iter().flatten().collect();
-    if record_batches.is_empty() {
+    let record_batches: Vec<RecordBatch> = index_record_batches.into_iter().flatten().collect();
+    if record_batches.is_empty() || record_batches.iter().all(|b| b.num_rows() == 0) {
         return Ok(());
     }
     let idx_schema: SchemaRef = record_batches.first().unwrap().schema();
@@ -653,7 +653,10 @@ pub(crate) async fn generate_index_on_ingester(
     let schema_key = idx_schema.hash_key();
     let schema_key_str = schema_key.as_str();
 
-    let json_rows = record_batches_to_json_rows(&record_batches)?;
+    let json_rows = record_batches_to_json_rows(&record_batches.iter().collect::<Vec<_>>())?;
+    if json_rows.is_empty() {
+        return Ok(());
+    }
     let recs: Vec<json::Value> = json_rows.into_iter().map(json::Value::Object).collect();
     for record_val in recs {
         let timestamp: i64 = record_val
@@ -698,13 +701,13 @@ pub(crate) async fn generate_index_on_compactor(
     org_id: &str,
     stream_name: &str,
 ) -> Result<(String, FileMeta), anyhow::Error> {
-    let mut index_record_batches =
+    let index_record_batches =
         prepare_index_record_batches_v1(buf, org_id, stream_name, &new_file_key).await?;
-    let schema = if let Some(first_batch) = index_record_batches.first() {
-        first_batch[0].schema()
-    } else {
+    let mut record_batches: Vec<RecordBatch> = index_record_batches.into_iter().flatten().collect();
+    if record_batches.is_empty() || record_batches.iter().all(|b| b.num_rows() == 0) {
         return Ok((String::new(), FileMeta::default()));
-    };
+    }
+    let schema = record_batches.first().unwrap().schema();
 
     let prefix_to_remove = format!("files/{}/logs/{}/", org_id, stream_name);
     let len_of_columns_to_invalidate = file_list_to_invalidate.len();
@@ -731,13 +734,11 @@ pub(crate) async fn generate_index_on_compactor(
     let deleted: ArrayRef = Arc::new(BooleanArray::from(vec![true; len_of_columns_to_invalidate]));
     let columns = vec![empty_terms, file_names, _timestamp, count, deleted];
     let batch = RecordBatch::try_new(schema, columns).unwrap();
-    index_record_batches.push(vec![batch]);
-
-    let record_batches_flattened = index_record_batches.into_iter().flatten().collect();
+    record_batches.push(batch);
 
     let original_file_size = 0; // The file never existed before this function was called
     let (filename, filemeta, _stream_type) = write_to_disk(
-        record_batches_flattened,
+        record_batches,
         original_file_size,
         org_id,
         stream_name,
@@ -760,14 +761,33 @@ async fn prepare_index_record_batches_v1(
     if batches.is_empty() {
         return Ok(vec![]);
     }
-    let local = batches.first().unwrap().schema();
+
+    // filter null columns
+    let schema = batches.first().unwrap().schema();
+    let batches = batches.iter().collect::<Vec<_>>();
+    let mut new_batch = arrow::compute::concat_batches(&schema, batches)?;
+    let mut null_columns = 0;
+    for i in 0..new_batch.num_columns() {
+        let ni = i - null_columns;
+        if new_batch.column(ni).null_count() == new_batch.num_rows() {
+            new_batch.remove_column(ni);
+            null_columns += 1;
+        }
+    }
+    let schema = new_batch.schema();
+    if schema.fields().is_empty()
+        || (schema.fields().len() == 1 && schema.field(0).name() == &CONFIG.common.column_timestamp)
+    {
+        return Ok(vec![]);
+    }
+
     let ctx = SessionContext::new();
-    let provider = MemTable::try_new(local.clone(), vec![batches])?;
+    let provider = MemTable::try_new(schema.clone(), vec![vec![new_batch]])?;
     ctx.register_table("_tbl_raw_data", Arc::new(provider))?;
     let prefix_to_remove = format!("files/{}/logs/{}/", org_id, stream_name);
     let file_name_without_prefix = new_file_key.trim_start_matches(&prefix_to_remove);
     let mut indexed_record_batches_to_merge = Vec::new();
-    for column in local.fields().iter() {
+    for column in schema.fields().iter() {
         if column.data_type() != &DataType::Utf8 {
             continue;
         }

src/service/logs/multi.rs

@@ -139,7 +139,7 @@ async fn ingest_inner(
                 &local_trans,
                 value,
                 &stream_vrl_map,
-                org_id, 
+                org_id,
                 stream_name,
                 &mut runtime,
             )?;

src/service/schema.rs

@@ -100,19 +100,12 @@ pub async fn check_for_schema(
         let (is_schema_changed, field_datatype_delta) =
             get_schema_changes(schema, &inferred_schema);
         if !is_schema_changed {
-            // generate new schema
-            let inferred_schema = if field_datatype_delta.is_empty() {
-                inferred_schema
-            } else {
-                inferred_schema.cloned_from(schema.schema())
-            };
             // check defined_schema_fields
             let stream_setting = get_settings(org_id, stream_name, stream_type).await;
             let defined_schema_fields = stream_setting
                 .and_then(|s| s.defined_schema_fields)
                 .unwrap_or_default();
             if !defined_schema_fields.is_empty() {
-                let schema = SchemaCache::new(inferred_schema);
                 let schema =
                     generate_schema_for_defined_schema_fields(schema, &defined_schema_fields);
                 stream_schema_map.insert(stream_name.to_string(), schema);
@@ -272,7 +265,7 @@ async fn handle_diff_schema(
         .unwrap_or_default();
     let final_schema = SchemaCache::new(final_schema);
     let final_schema =
-        generate_schema_for_defined_schema_fields(final_schema, &defined_schema_fields);
+        generate_schema_for_defined_schema_fields(&final_schema, &defined_schema_fields);
 
     // update thread cache
     stream_schema_map.insert(stream_name.to_string(), final_schema);
@@ -287,11 +280,11 @@ async fn handle_diff_schema(
 // if defined_schema_fields is not empty, and schema fields greater than defined_schema_fields + 10,
 // then we will use defined_schema_fields
 fn generate_schema_for_defined_schema_fields(
-    schema: SchemaCache,
+    schema: &SchemaCache,
     fields: &[String],
 ) -> SchemaCache {
     if fields.is_empty() || schema.fields_map().len() < fields.len() + 10 {
-        return schema;
+        return schema.clone();
     }
 
     let mut fields: HashSet<_> = fields.iter().collect();