fix: metrics Mismatch between schema and batches (#5673)

This commit is contained in:
Hengfei Yang
2025-01-12 15:13:00 +08:00
committed by GitHub
parent 36de81c437
commit 445a50ddde
2 changed files with 32 additions and 5 deletions
+8
View File
@@ -352,6 +352,10 @@ pub async fn merge(
schema: &Schema,
min_ts: Option<i64>,
) -> Result<Option<(Schema, Vec<Field>)>> {
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<String, String>,
) -> 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(
+24 -5
View File
@@ -13,7 +13,10 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
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))
}