diff --git a/src/config/src/utils/arrow.rs b/src/config/src/utils/arrow.rs index d81b4cb464..76c4fcc357 100644 --- a/src/config/src/utils/arrow.rs +++ b/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>, 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() diff --git a/src/infra/src/schema/mod.rs b/src/infra/src/schema/mod.rs index 1805a8c2d1..1e6360126c 100644 --- a/src/infra/src/schema/mod.rs +++ b/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())) } diff --git a/src/job/files/parquet.rs b/src/job/files/parquet.rs index 2c0e22c2ed..a37d3b898b 100644 --- a/src/job/files/parquet.rs +++ b/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 = 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::>())?; + if json_rows.is_empty() { + return Ok(()); + } let recs: Vec = 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 = 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::>(); + 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; } diff --git a/src/service/logs/multi.rs b/src/service/logs/multi.rs index 52ecfd5bff..f7473399e8 100644 --- a/src/service/logs/multi.rs +++ b/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, )?; diff --git a/src/service/schema.rs b/src/service/schema.rs index 405acc8c44..bc3d0b0232 100644 --- a/src/service/schema.rs +++ b/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();