diff --git a/src/infra/src/schema/mod.rs b/src/infra/src/schema/mod.rs index 3b0645fbf2..b7e9fa97c7 100644 --- a/src/infra/src/schema/mod.rs +++ b/src/infra/src/schema/mod.rs @@ -352,6 +352,10 @@ pub async fn merge( schema: &Schema, min_ts: Option, ) -> Result)>> { + let stream_name = stream_name.trim(); + if stream_name.is_empty() { + return Ok(None); + } let start_dt = min_ts; let key = mk_key(org_id, stream_type, stream_name); let inferred_schema = schema.clone(); @@ -465,6 +469,10 @@ pub async fn update_setting( stream_type: StreamType, metadata: std::collections::HashMap, ) -> Result<()> { + let stream_name = stream_name.trim(); + if stream_name.is_empty() { + return Ok(()); + } let key = mk_key(org_id, stream_type, stream_name); let db = infra_db::get_db().await; db.get_for_update( diff --git a/src/service/promql/search/grpc/wal.rs b/src/service/promql/search/grpc/wal.rs index 9bfdb2eab5..c9e035ba4f 100644 --- a/src/service/promql/search/grpc/wal.rs +++ b/src/service/promql/search/grpc/wal.rs @@ -13,7 +13,10 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -use std::{collections::HashSet, sync::Arc}; +use std::{ + collections::{HashMap, HashSet}, + sync::Arc, +}; use arrow::record_batch::RecordBatch; use config::{ @@ -97,6 +100,7 @@ pub(crate) async fn create_context( let ctx = prepare_datafusion_context(None, vec![], false, 0).await?; let mem_table = Arc::new(MemTable::try_new(schema.clone(), vec![batches])?); + log::info!("[trace_id {trace_id}] promql->wal->search: register mem table done"); ctx.register_table(stream_name, mem_table)?; resp.push((ctx, schema, stats)); @@ -189,9 +193,24 @@ async fn get_wal_batches( ret.map(|data| (data, visit.scan_stats, visit.partial_err)) }?; - let mut schema = Arc::new(Schema::empty()); - if !batches.is_empty() { - schema = Arc::clone(&batches[0].schema()); + if batches.is_empty() { + return Ok((ScanStats::new(), vec![], Arc::new(Schema::empty()))); } - Ok((stats, batches, schema)) + + // remove the metadata + let schema = Arc::new( + batches[0] + .schema() + .as_ref() + .clone() + .with_metadata(HashMap::new()), + ); + let mut new_batches = Vec::with_capacity(batches.len()); + for batch in batches { + new_batches.push(RecordBatch::try_new( + schema.clone(), + batch.columns().to_vec(), + )?); + } + Ok((stats, new_batches, schema)) }