From a906e6e9dd4097c5e12514d6272a330d9b83539b Mon Sep 17 00:00:00 2001 From: "nadav.govari" Date: Wed, 30 Sep 2026 11:19:08 -0400 Subject: [PATCH 1/2] Track ingester shard queue size --- quickwit/Cargo.lock | 2 +- quickwit/Cargo.toml | 2 +- .../quickwit-ingest/src/ingest_v2/ingester.rs | 20 ++++++-- .../quickwit-ingest/src/ingest_v2/models.rs | 10 ++++ .../src/ingest_v2/mrecordlog_utils.rs | 46 ++++++++++++++++++- .../quickwit-ingest/src/ingest_v2/state.rs | 16 ++++++- .../quickwit-ingest/src/mrecordlog_async.rs | 9 +++- 7 files changed, 95 insertions(+), 10 deletions(-) diff --git a/quickwit/Cargo.lock b/quickwit/Cargo.lock index 06f60aafc63..440f5111786 100644 --- a/quickwit/Cargo.lock +++ b/quickwit/Cargo.lock @@ -5213,7 +5213,7 @@ dependencies = [ [[package]] name = "mrecordlog" version = "0.5.0" -source = "git+https://github.com/quickwit-oss/mrecordlog?rev=3b3562ef#3b3562efea468a3a72a244b16d3c555db0ddc95b" +source = "git+https://github.com/quickwit-oss/mrecordlog?rev=e79bed2dbe4077f02479d073574c9f7a7ab03b85#e79bed2dbe4077f02479d073574c9f7a7ab03b85" dependencies = [ "bytes", "crc32fast", diff --git a/quickwit/Cargo.toml b/quickwit/Cargo.toml index e8f90aad546..e6a600c255a 100644 --- a/quickwit/Cargo.toml +++ b/quickwit/Cargo.toml @@ -133,7 +133,7 @@ metrics-util = "0.20" mime_guess = "2.0" mini-moka = "0.10" mockall = "0.14" -mrecordlog = { git = "https://github.com/quickwit-oss/mrecordlog", rev = "3b3562ef" } +mrecordlog = { git = "https://github.com/quickwit-oss/mrecordlog", rev = "e79bed2dbe4077f02479d073574c9f7a7ab03b85" } new_string_template = "1.5" nom = "8.0" numfmt = "1.2" diff --git a/quickwit/quickwit-ingest/src/ingest_v2/ingester.rs b/quickwit/quickwit-ingest/src/ingest_v2/ingester.rs index 3d6e6bfa453..ee7ec5fc001 100644 --- a/quickwit/quickwit-ingest/src/ingest_v2/ingester.rs +++ b/quickwit/quickwit-ingest/src/ingest_v2/ingester.rs @@ -53,7 +53,8 @@ use super::local_shards::ShardThroughputReadings; use super::metrics::report_local_shards_metrics; use super::models::IngesterShard; use super::mrecordlog_utils::{ - AppendDocBatchError, append_non_empty_doc_batch, check_enough_capacity, wal_stats, + AppendDocBatchError, append_non_empty_doc_batch, check_enough_capacity, doc_batch_size, + wal_stats, }; use super::rate_meter::RateMeter; use super::state::{IngesterState, InnerIngesterState, WeakIngesterState}; @@ -597,6 +598,7 @@ impl Ingester { let queue_id = subrequest.queue_id; let batch_num_docs = subrequest.doc_batch.num_docs() as u64; + let batch_size = doc_batch_size(&subrequest.doc_batch, force_commit); let append_result = append_non_empty_doc_batch( &mut state_guard.mrecordlog, @@ -637,11 +639,12 @@ impl Ingester { } }; - state_guard + let shard = state_guard .shards .get_mut(&queue_id) - .expect("shard should exist") - .set_replication_position_inclusive(current_position_inclusive.clone(), now); + .expect("shard should exist"); + shard.set_replication_position_inclusive(current_position_inclusive.clone(), now); + shard.queue_size += batch_size; let persist_success = PersistSuccess { subrequest_id: subrequest.subrequest_id, @@ -1199,6 +1202,7 @@ mod tests { use crate::ingest_v2::DEFAULT_IDLE_SHARD_TIMEOUT; use crate::ingest_v2::doc_mapper::try_build_doc_mapper; use crate::ingest_v2::fetch::tests::{into_fetch_eof, into_fetch_payload}; + use crate::ingest_v2::mrecordlog_utils::read_queue_size; pub(super) struct IngesterForTest { node_id: NodeId, @@ -1782,6 +1786,10 @@ mod tests { let shard_01 = state_guard.shards.get(&queue_id_01).unwrap(); shard_01.assert_is_open(); shard_01.assert_replication_position(Position::offset(1u64)); + assert_eq!( + shard_01.queue_size, + read_queue_size(&state_guard.mrecordlog, &queue_id_01) + ); state_guard.mrecordlog.assert_records_eq( &queue_id_01, @@ -1793,6 +1801,10 @@ mod tests { let shard_11 = state_guard.shards.get(&queue_id_11).unwrap(); shard_11.assert_is_open(); shard_11.assert_replication_position(Position::offset(2u64)); + assert_eq!( + shard_11.queue_size, + read_queue_size(&state_guard.mrecordlog, &queue_id_11) + ); state_guard.mrecordlog.assert_records_eq( &queue_id_11, diff --git a/quickwit/quickwit-ingest/src/ingest_v2/models.rs b/quickwit/quickwit-ingest/src/ingest_v2/models.rs index cd6fa2d3dd5..91dbd1a5882 100644 --- a/quickwit/quickwit-ingest/src/ingest_v2/models.rs +++ b/quickwit/quickwit-ingest/src/ingest_v2/models.rs @@ -15,6 +15,7 @@ use std::sync::Arc; use std::time::{Duration, Instant}; +use bytesize::ByteSize; use quickwit_common::rate_limiter::RateLimiter; use quickwit_doc_mapper::DocMapper; use quickwit_proto::ingest::ShardState; @@ -37,6 +38,7 @@ pub(super) struct IngesterShard { pub replication_position_inclusive: Position, /// Position up to which the shard has been truncated. pub truncation_position_inclusive: Position, + pub queue_size: ByteSize, pub rate_limiter: RateLimiter, pub rate_meter: RateMeter, /// Whether the shard should be advertised to other nodes (routers) via gossip. @@ -67,6 +69,7 @@ pub(super) struct IngesterShardBuilder { shard_state: ShardState, replication_position_inclusive: Position, truncation_position_inclusive: Position, + queue_size: ByteSize, rate_limiter: RateLimiter, rate_meter: RateMeter, doc_mapper_opt: Option>, @@ -112,6 +115,11 @@ impl IngesterShardBuilder { self } + pub fn with_queue_size(mut self, queue_size: ByteSize) -> Self { + self.queue_size = queue_size; + self + } + /// Sets whether to validate documents. Defaults to `false`. pub fn with_validate_docs(mut self, validate_docs: bool) -> Self { self.validate_docs = validate_docs; @@ -144,6 +152,7 @@ impl IngesterShardBuilder { shard_state: self.shard_state, replication_position_inclusive: self.replication_position_inclusive, truncation_position_inclusive: self.truncation_position_inclusive, + queue_size: self.queue_size, rate_limiter: self.rate_limiter, rate_meter: self.rate_meter, is_advertisable: self.is_advertisable, @@ -170,6 +179,7 @@ impl IngesterShard { shard_state: ShardState::Open, replication_position_inclusive: Position::Beginning, truncation_position_inclusive: Position::Beginning, + queue_size: ByteSize::default(), rate_limiter: RateLimiter::default(), rate_meter: RateMeter::default(), doc_mapper_opt: None, diff --git a/quickwit/quickwit-ingest/src/ingest_v2/mrecordlog_utils.rs b/quickwit/quickwit-ingest/src/ingest_v2/mrecordlog_utils.rs index 16c802e9237..8a7b039053a 100644 --- a/quickwit/quickwit-ingest/src/ingest_v2/mrecordlog_utils.rs +++ b/quickwit/quickwit-ingest/src/ingest_v2/mrecordlog_utils.rs @@ -23,8 +23,9 @@ use quickwit_proto::ingest::DocBatchV2; use quickwit_proto::types::{Position, QueueId}; use tracing::instrument; -use crate::MRecord; +use super::mrecord::MRECORD_HEADER_LEN; use crate::mrecordlog_async::MultiRecordLogAsync; +use crate::{MRecord, estimate_size}; #[derive(Debug, thiserror::Error)] pub(super) enum AppendDocBatchError { @@ -98,6 +99,24 @@ pub(super) async fn append_non_empty_doc_batch( } } +pub(super) fn doc_batch_size(doc_batch: &DocBatchV2, force_commit: bool) -> ByteSize { + let estimated_size = estimate_size(doc_batch); + if force_commit { + estimated_size + ByteSize::b(MRECORD_HEADER_LEN as u64) + } else { + estimated_size + } +} + +pub(super) fn read_queue_size(mrecordlog: &MultiRecordLogAsync, queue_id: &QueueId) -> ByteSize { + let num_bytes: usize = mrecordlog + .range(queue_id, ..) + .expect("queue should exist") + .map(|record| record.payload.len()) + .sum(); + ByteSize::b(num_bytes as u64) +} + /// Error returned when the mrecordlog does not have enough capacity to store some records. #[derive(Debug, Clone, Copy, thiserror::Error)] pub(super) enum NotEnoughCapacityError { @@ -210,6 +229,31 @@ mod tests { assert_eq!(position, Position::offset(2u64)); } + #[cfg(not(feature = "failpoints"))] + #[tokio::test] + async fn test_doc_batch_size_and_read_queue_size() { + let tempdir = tempfile::tempdir().unwrap(); + let mut mrecordlog = MultiRecordLogAsync::open(tempdir.path()).await.unwrap(); + + let queue_id = "test-queue".to_string(); + mrecordlog.create_queue(&queue_id).await.unwrap(); + assert_eq!(read_queue_size(&mrecordlog, &queue_id), ByteSize::b(0)); + + let doc_batch = DocBatchV2::for_test(["test-doc-foo", "test-doc-bar"]); + assert_eq!(doc_batch_size(&doc_batch, false), ByteSize::b(28)); + assert_eq!(doc_batch_size(&doc_batch, true), ByteSize::b(30)); + + append_non_empty_doc_batch(&mut mrecordlog, &queue_id, doc_batch.clone(), false) + .await + .unwrap(); + assert_eq!(read_queue_size(&mrecordlog, &queue_id), ByteSize::b(28)); + + append_non_empty_doc_batch(&mut mrecordlog, &queue_id, doc_batch, true) + .await + .unwrap(); + assert_eq!(read_queue_size(&mrecordlog, &queue_id), ByteSize::b(58)); + } + // This test should be run manually and independently of other tests with the `failpoints` // feature enabled: // ```sh diff --git a/quickwit/quickwit-ingest/src/ingest_v2/state.rs b/quickwit/quickwit-ingest/src/ingest_v2/state.rs index 3bcdbae1867..a8745d03102 100644 --- a/quickwit/quickwit-ingest/src/ingest_v2/state.rs +++ b/quickwit/quickwit-ingest/src/ingest_v2/state.rs @@ -42,6 +42,7 @@ use tracing::{error, info, instrument}; use super::local_shards::{ShardInfo, ShardInfos, ShardThroughputReadings}; use super::metrics::report_local_shards_metrics; use super::models::IngesterShard; +use super::mrecordlog_utils::read_queue_size; use super::rate_meter::RateMeter; use super::wal_capacity_tracker::WalCapacityTracker; use crate::OpenShardCounts; @@ -392,6 +393,7 @@ impl IngesterState { .checked_sub(1) .map(Position::offset) .unwrap_or(Position::Beginning); + let queue_size = read_queue_size(&mrecordlog, &queue_id); let rate_limiter = RateLimiter::from_settings(rate_limiter_settings); let rate_meter = RateMeter::default(); @@ -400,6 +402,7 @@ impl IngesterState { .with_state(ShardState::Closed) .with_replication_position_inclusive(replication_position_inclusive) .with_truncation_position_inclusive(truncation_position_inclusive) + .with_queue_size(queue_size) .with_rate_limiter(rate_limiter) .with_rate_meter(rate_meter) .with_last_write(now) @@ -714,7 +717,13 @@ impl FullyLockedIngesterState<'_> { .truncate(queue_id, truncate_up_to_offset_inclusive) .await { - Ok(_) => {} + Ok(evicted_size) => { + let queue_size = shard + .queue_size + .as_u64() + .saturating_sub(evicted_size.as_u64()); + shard.queue_size = ByteSize::b(queue_size); + } Err(TruncateError::MissingQueue(_)) => { error!("failed to truncate shard `{queue_id}`: WAL queue not found"); self.shards.remove(queue_id); @@ -1053,6 +1062,7 @@ mod tests { shard_01.truncation_position_inclusive, Position::offset(0u64) ); + assert_eq!(shard_01.queue_size, ByteSize::b(24)); // Fully truncated queue: recovers at its last position rather than the beginning. let shard_02 = state_guard.shards.get(&queue_id_02).unwrap(); @@ -1065,12 +1075,14 @@ mod tests { shard_02.truncation_position_inclusive, Position::offset(1u64) ); + assert_eq!(shard_02.queue_size, ByteSize::b(0)); // Never-written queue: recovers at the beginning. let shard_03 = state_guard.shards.get(&queue_id_03).unwrap(); assert_eq!(shard_03.shard_state, ShardState::Closed); assert_eq!(shard_03.replication_position_inclusive, Position::Beginning); assert_eq!(shard_03.truncation_position_inclusive, Position::Beginning); + assert_eq!(shard_03.queue_size, ByteSize::b(0)); } fn insert_shard_with_used_capacity( @@ -1361,6 +1373,7 @@ mod tests { IngesterShard::builder(index_uid.clone(), source_id.clone(), ShardId::from(2)) .with_state(ShardState::Closed) .with_replication_position_inclusive(Position::offset(1u64)) + .with_queue_size(ByteSize::b(24)) .build(); state_guard.shards.insert(queue_id_02.clone(), shard_02); @@ -1387,6 +1400,7 @@ mod tests { .await; let shard_02 = state_guard.shards.get(&queue_id_02).unwrap(); assert_eq!(shard_02.truncation_position_inclusive, Position::eof(1u64)); + assert_eq!(shard_02.queue_size, ByteSize::b(0)); state_guard .mrecordlog .assert_records_eq(&queue_id_02, .., &[]); diff --git a/quickwit/quickwit-ingest/src/mrecordlog_async.rs b/quickwit/quickwit-ingest/src/mrecordlog_async.rs index 4fc0f8aea04..fb83d167f2f 100644 --- a/quickwit/quickwit-ingest/src/mrecordlog_async.rs +++ b/quickwit/quickwit-ingest/src/mrecordlog_async.rs @@ -17,6 +17,7 @@ use std::ops::RangeBounds; use std::path::Path; use bytes::Buf; +use bytesize::ByteSize; use mrecordlog::error::*; use mrecordlog::{MultiRecordLog, PersistAction, PersistPolicy, Record, ResourceUsage}; use tokio::task::JoinError; @@ -151,7 +152,11 @@ impl MultiRecordLogAsync { } #[instrument(name = "mrecordlog.truncate_async", skip_all, fields(queue, position))] - pub async fn truncate(&mut self, queue: &str, position: u64) -> Result { + pub async fn truncate( + &mut self, + queue: &str, + position: u64, + ) -> Result { let span = info_span!("mrecordlog.truncate", queue, position); let queue = queue.to_string(); self.run_operation(span, move |mrecordlog| { @@ -160,7 +165,7 @@ impl MultiRecordLogAsync { .inspect(|outcome| { WAL_BYTES_WRITTEN_TRUNCATE.inc_by(outcome.wal_bytes_written); }) - .map(|outcome| outcome.evicted_records) + .map(|outcome| ByteSize::b(outcome.evicted_bytes as u64)) }) .await } From 1e17f3b6e29d74720c0dadd4acc73c0e40aa9b2c Mon Sep 17 00:00:00 2001 From: "nadav.govari" Date: Wed, 7 Oct 2026 10:42:58 -0400 Subject: [PATCH 2/2] PR comments: use updated mrecordlog implementation --- quickwit/Cargo.lock | 2 +- quickwit/Cargo.toml | 2 +- .../src/source/ingest/mod.rs | 8 +- .../src/ingest_v2/doc_mapper.rs | 2 +- .../quickwit-ingest/src/ingest_v2/fetch.rs | 4 +- .../quickwit-ingest/src/ingest_v2/ingester.rs | 108 +++++++++++++++--- quickwit/quickwit-ingest/src/ingest_v2/mod.rs | 35 ++---- .../src/ingest_v2/mrecordlog_utils.rs | 88 ++++++++------ .../quickwit-ingest/src/ingest_v2/state.rs | 11 +- .../quickwit-ingest/src/mrecordlog_async.rs | 15 +-- quickwit/quickwit-ingest/src/queue.rs | 4 +- .../quickwit-opentelemetry/src/otlp/logs.rs | 6 +- .../quickwit-opentelemetry/src/otlp/traces.rs | 6 +- .../quickwit-proto/src/ingest/ingester.rs | 4 +- quickwit/quickwit-proto/src/ingest/mod.rs | 6 +- .../src/elasticsearch_api/bulk_v2.rs | 15 ++- 16 files changed, 197 insertions(+), 119 deletions(-) diff --git a/quickwit/Cargo.lock b/quickwit/Cargo.lock index 440f5111786..d69715c8d36 100644 --- a/quickwit/Cargo.lock +++ b/quickwit/Cargo.lock @@ -5213,7 +5213,7 @@ dependencies = [ [[package]] name = "mrecordlog" version = "0.5.0" -source = "git+https://github.com/quickwit-oss/mrecordlog?rev=e79bed2dbe4077f02479d073574c9f7a7ab03b85#e79bed2dbe4077f02479d073574c9f7a7ab03b85" +source = "git+https://github.com/quickwit-oss/mrecordlog?rev=582ddb131181a053635e6904d3190491a3ee556c#582ddb131181a053635e6904d3190491a3ee556c" dependencies = [ "bytes", "crc32fast", diff --git a/quickwit/Cargo.toml b/quickwit/Cargo.toml index e6a600c255a..2acc7bf737a 100644 --- a/quickwit/Cargo.toml +++ b/quickwit/Cargo.toml @@ -133,7 +133,7 @@ metrics-util = "0.20" mime_guess = "2.0" mini-moka = "0.10" mockall = "0.14" -mrecordlog = { git = "https://github.com/quickwit-oss/mrecordlog", rev = "e79bed2dbe4077f02479d073574c9f7a7ab03b85" } +mrecordlog = { git = "https://github.com/quickwit-oss/mrecordlog", rev = "582ddb131181a053635e6904d3190491a3ee556c" } new_string_template = "1.5" nom = "8.0" numfmt = "1.2" diff --git a/quickwit/quickwit-indexing/src/source/ingest/mod.rs b/quickwit/quickwit-indexing/src/source/ingest/mod.rs index ba27af469b2..1c9cd493693 100644 --- a/quickwit/quickwit-indexing/src/source/ingest/mod.rs +++ b/quickwit/quickwit-indexing/src/source/ingest/mod.rs @@ -1560,7 +1560,7 @@ mod tests { from_position_exclusive: Some(Position::offset(11u64)), to_position_inclusive: Some(Position::offset(14u64)), }; - let batch_size = fetch_payload.estimate_size(); + let batch_size = fetch_payload.buffer_size(); let fetch_message = FetchMessage::new_payload(fetch_payload); let in_flight_value = InFlightValue::new(fetch_message, batch_size, &IN_FLIGHT_FETCH_STREAM); @@ -1574,7 +1574,7 @@ mod tests { from_position_exclusive: Some(Position::offset(22u64)), to_position_inclusive: Some(Position::offset(23u64)), }; - let batch_size = fetch_payload.estimate_size(); + let batch_size = fetch_payload.buffer_size(); let fetch_message = FetchMessage::new_payload(fetch_payload); let in_flight_value = InFlightValue::new(fetch_message, batch_size, &IN_FLIGHT_FETCH_STREAM); @@ -1653,7 +1653,7 @@ mod tests { from_position_exclusive: Some(Position::offset(14u64)), to_position_inclusive: Some(Position::offset(15u64)), }; - let batch_size = fetch_payload.estimate_size(); + let batch_size = fetch_payload.buffer_size(); let fetch_message = FetchMessage::new_payload(fetch_payload); let in_flight_value = InFlightValue::new(fetch_message, batch_size, &IN_FLIGHT_FETCH_STREAM); @@ -1795,7 +1795,7 @@ mod tests { from_position_exclusive: Some(Position::offset(11u64)), to_position_inclusive: Some(Position::offset(13u64)), }; - let batch_size = fetch_payload.estimate_size(); + let batch_size = fetch_payload.buffer_size(); let fetch_message = FetchMessage::new_payload(fetch_payload); let in_flight_value = InFlightValue::new(fetch_message, batch_size, &IN_FLIGHT_FETCH_STREAM); diff --git a/quickwit/quickwit-ingest/src/ingest_v2/doc_mapper.rs b/quickwit/quickwit-ingest/src/ingest_v2/doc_mapper.rs index 0ae0cb536b6..b1b7727bb0a 100644 --- a/quickwit/quickwit-ingest/src/ingest_v2/doc_mapper.rs +++ b/quickwit/quickwit-ingest/src/ingest_v2/doc_mapper.rs @@ -85,7 +85,7 @@ fn is_document_validation_enabled() -> bool { !quickwit_common::get_bool_from_env_cached!("QW_DISABLE_DOCUMENT_VALIDATION", false) } -#[instrument(name = "ingester.validate_doc_batch", skip_all, fields(num_docs = doc_batch.num_docs(), num_bytes = doc_batch.num_bytes()))] +#[instrument(name = "ingester.validate_doc_batch", skip_all, fields(num_docs = doc_batch.num_docs(), num_bytes = doc_batch.doc_buffer.len()))] async fn validate_doc_batch_cpu_intensive( doc_batch: DocBatchV2, doc_mapper: Arc, diff --git a/quickwit/quickwit-ingest/src/ingest_v2/fetch.rs b/quickwit/quickwit-ingest/src/ingest_v2/fetch.rs index 762e1ab143b..bfae1f43c17 100644 --- a/quickwit/quickwit-ingest/src/ingest_v2/fetch.rs +++ b/quickwit/quickwit-ingest/src/ingest_v2/fetch.rs @@ -170,7 +170,7 @@ impl FetchStreamTask { mrecord_buffer: mrecord_buffer.freeze(), mrecord_lengths, }; - let batch_size = mrecord_batch.estimate_size(); + let batch_size = mrecord_batch.buffer_size(); let fetch_payload = FetchPayload { index_uid: Some(self.index_uid.clone()), source_id: self.source_id.clone(), @@ -490,7 +490,7 @@ async fn fetch_stream_once( match fetch_message_result { Ok(fetch_message) => match &fetch_message.message { Some(fetch_message::Message::Payload(fetch_payload)) => { - let batch_size = fetch_payload.estimate_size(); + let batch_size = fetch_payload.buffer_size(); let to_position_inclusive = fetch_payload.to_position_inclusive(); let in_flight_value = InFlightValue::new( fetch_message, diff --git a/quickwit/quickwit-ingest/src/ingest_v2/ingester.rs b/quickwit/quickwit-ingest/src/ingest_v2/ingester.rs index ee7ec5fc001..e3675f0b397 100644 --- a/quickwit/quickwit-ingest/src/ingest_v2/ingester.rs +++ b/quickwit/quickwit-ingest/src/ingest_v2/ingester.rs @@ -58,7 +58,6 @@ use super::mrecordlog_utils::{ }; use super::rate_meter::RateMeter; use super::state::{IngesterState, InnerIngesterState, WeakIngesterState}; -use crate::estimate_size; use crate::ingest_v2::doc_mapper::get_or_try_build_doc_mapper; use crate::ingest_v2::metrics::{ DECOMMISSION_FAILED, DECOMMISSION_SUCCEEDED, RESET_SHARDS_OPERATIONS_TOTAL, STATUS, @@ -477,7 +476,7 @@ impl Ingester { DocBatchV2::default() } }; - let requested_capacity = estimate_size(&doc_batch); + let requested_capacity = doc_batch_size(&doc_batch, force_commit); if let Err(error) = check_enough_capacity( &state_guard.mrecordlog, @@ -518,7 +517,7 @@ impl Ingester { } // Total number of bytes (valid and invalid documents) - let original_batch_num_bytes = doc_batch.num_bytes() as u64; + let original_batch_num_bytes = doc_batch.doc_buffer.len() as u64; let (valid_doc_batch, parse_failures) = if validate_docs { validate_doc_batch(doc_batch, doc_mapper).await? @@ -559,7 +558,7 @@ impl Ingester { parent: DOCS_BYTES_TOTAL, labels: [label_values!(VALIDITY => "valid")], ) - .inc_by(valid_doc_batch.num_bytes() as u64); + .inc_by(valid_doc_batch.doc_buffer.len() as u64); if !parse_failures.is_empty() { counter!( parent: DOCS_TOTAL, @@ -570,9 +569,9 @@ impl Ingester { parent: DOCS_BYTES_TOTAL, labels: [label_values!(VALIDITY => "invalid")], ) - .inc_by(original_batch_num_bytes - valid_doc_batch.num_bytes() as u64); + .inc_by(original_batch_num_bytes - valid_doc_batch.doc_buffer.len() as u64); } - let valid_batch_num_bytes = valid_doc_batch.num_bytes() as u64; + let valid_batch_num_bytes = valid_doc_batch.doc_buffer.len() as u64; shard.rate_meter.update(valid_batch_num_bytes); total_requested_capacity += requested_capacity; @@ -598,7 +597,6 @@ impl Ingester { let queue_id = subrequest.queue_id; let batch_num_docs = subrequest.doc_batch.num_docs() as u64; - let batch_size = doc_batch_size(&subrequest.doc_batch, force_commit); let append_result = append_non_empty_doc_batch( &mut state_guard.mrecordlog, @@ -608,8 +606,8 @@ impl Ingester { ) .await; - let current_position_inclusive = match append_result { - Ok(current_position_inclusive) => current_position_inclusive, + let (current_position_inclusive, queue_size) = match append_result { + Ok(outcome) => outcome, Err(append_error) => { let reason = match &append_error { AppendDocBatchError::Io(io_error) => { @@ -644,7 +642,7 @@ impl Ingester { .get_mut(&queue_id) .expect("shard should exist"); shard.set_replication_position_inclusive(current_position_inclusive.clone(), now); - shard.queue_size += batch_size; + shard.queue_size = queue_size; let persist_success = PersistSuccess { subrequest_id: subrequest.subrequest_id, @@ -928,7 +926,9 @@ impl IngesterService for Ingester { .subrequests .iter() .flat_map(|subrequest| match &subrequest.doc_batch { - Some(doc_batch) if doc_batch.doc_buffer.is_unique() => Some(doc_batch.num_bytes()), + Some(doc_batch) if doc_batch.doc_buffer.is_unique() => { + Some(doc_batch.doc_buffer.len()) + } _ => None, }) .sum::(); @@ -1202,7 +1202,6 @@ mod tests { use crate::ingest_v2::DEFAULT_IDLE_SHARD_TIMEOUT; use crate::ingest_v2::doc_mapper::try_build_doc_mapper; use crate::ingest_v2::fetch::tests::{into_fetch_eof, into_fetch_payload}; - use crate::ingest_v2::mrecordlog_utils::read_queue_size; pub(super) struct IngesterForTest { node_id: NodeId, @@ -1687,6 +1686,87 @@ mod tests { assert!(shard.long_term_ingestion_rate.as_u64() > 0); } + #[tokio::test] + async fn test_queue_size_does_not_drift_after_persist_and_truncate() { + let (ingester_ctx, ingester) = IngesterForTest::default().build().await; + let index_uid = IndexUid::for_test("test-index", 0); + let source_id = SourceId::from("test-source"); + let doc_mapping_uid = DocMappingUid::random(); + let response = ingester + .init_shards(InitShardsRequest { + subrequests: vec![InitShardSubrequest { + subrequest_id: 0, + shard: Some(Shard { + index_uid: Some(index_uid.clone()), + source_id: source_id.clone(), + shard_id: Some(ShardId::from(1)), + shard_state: ShardState::Open as i32, + ingester_id: ingester_ctx.node_id.to_string(), + doc_mapping_uid: Some(doc_mapping_uid), + ..Default::default() + }), + doc_mapping_json: format!(r#"{{"doc_mapping_uid":"{doc_mapping_uid}"}}"#), + validate_docs: false, + }], + }) + .await + .unwrap(); + assert_eq!(response.successes.len(), 1); + assert!(response.failures.is_empty()); + + let queue_id = queue_id(&index_uid, &source_id, &ShardId::from(1)); + let mut next_position = 0u64; + let mut retained_bytes = 0u64; + for iteration in 0..32 { + let force_commit = iteration % 2 == 0; + let commit_type = if force_commit { + CommitTypeV2::Force + } else { + CommitTypeV2::Auto + }; + let response = ingester + .persist(PersistRequest { + ingester_id: ingester_ctx.node_id.to_string(), + commit_type: commit_type as i32, + subrequests: vec![PersistSubrequest { + subrequest_id: 0, + index_uid: Some(index_uid.clone()), + source_id: source_id.clone(), + doc_batch: Some(DocBatchV2::for_test(["a", "longer", "last"])), + }], + }) + .await + .unwrap(); + assert_eq!(response.successes.len(), 1); + assert!(response.failures.is_empty()); + + let mut state_guard = ingester.state.lock_fully("test").await.unwrap(); + let commit_bytes = u64::from(force_commit) * 2; + assert_eq!( + state_guard.shards[&queue_id].queue_size, + ByteSize::b(retained_bytes + 17 + commit_bytes) + ); + state_guard + .truncate_shard(&queue_id, Position::offset(next_position + 1), "test") + .await; + retained_bytes = 6 + commit_bytes; + assert_eq!( + state_guard.mrecordlog.summary().queues[&queue_id].num_bytes as u64, + retained_bytes + ); + assert_eq!( + state_guard.shards[&queue_id].queue_size, + ByteSize::b(retained_bytes) + ); + next_position += 3 + u64::from(force_commit); + } + let mut state_guard = ingester.state.lock_fully("test").await.unwrap(); + state_guard + .truncate_shard(&queue_id, Position::offset(next_position - 1), "test") + .await; + assert_eq!(state_guard.shards[&queue_id].queue_size, ByteSize::b(0)); + } + #[tokio::test] async fn test_ingester_persist() { let (ingester_ctx, ingester) = IngesterForTest::default().build().await; @@ -1788,7 +1868,7 @@ mod tests { shard_01.assert_replication_position(Position::offset(1u64)); assert_eq!( shard_01.queue_size, - read_queue_size(&state_guard.mrecordlog, &queue_id_01) + ByteSize::b(state_guard.mrecordlog.summary().queues[&queue_id_01].num_bytes as u64) ); state_guard.mrecordlog.assert_records_eq( @@ -1803,7 +1883,7 @@ mod tests { shard_11.assert_replication_position(Position::offset(2u64)); assert_eq!( shard_11.queue_size, - read_queue_size(&state_guard.mrecordlog, &queue_id_11) + ByteSize::b(state_guard.mrecordlog.summary().queues[&queue_id_11].num_bytes as u64) ); state_guard.mrecordlog.assert_records_eq( diff --git a/quickwit/quickwit-ingest/src/ingest_v2/mod.rs b/quickwit/quickwit-ingest/src/ingest_v2/mod.rs index 7e00eff563e..7316c7c6df4 100644 --- a/quickwit/quickwit-ingest/src/ingest_v2/mod.rs +++ b/quickwit/quickwit-ingest/src/ingest_v2/mod.rs @@ -59,7 +59,6 @@ use workbench::pending_subrequests; pub use self::fetch::{FetchStreamError, MultiFetchStream}; pub use self::ingester::Ingester; -use self::mrecord::MRECORD_HEADER_LEN; pub use self::mrecord::{MRecord, decoded_mrecords}; pub use self::router::IngestRouter; @@ -280,11 +279,6 @@ impl IngestRequestV2Builder { } } -pub(super) fn estimate_size(doc_batch: &DocBatchV2) -> ByteSize { - let estimate = doc_batch.num_bytes() + doc_batch.num_docs() * MRECORD_HEADER_LEN; - ByteSize(estimate as u64) -} - #[derive(Debug, Clone, Copy, Default, Eq, PartialEq, Ord, PartialOrd)] pub struct RateMibPerSec(pub u16); @@ -335,7 +329,7 @@ mod tests { let doc_batch = doc_batch_builder.build().unwrap(); assert_eq!(doc_batch.num_docs(), 2); - assert_eq!(doc_batch.num_bytes(), 21); + assert_eq!(doc_batch.doc_buffer.len(), 13); assert_eq!(doc_batch.doc_lengths, [7, 6]); assert_eq!(doc_batch.doc_buffer, Bytes::from(&b"Hello, World!"[..])); } @@ -391,8 +385,9 @@ mod tests { .doc_batch .as_ref() .unwrap() - .num_bytes(), - 21 + .doc_buffer + .len(), + 13 ); assert_eq!( ingest_request.subrequests[0] @@ -434,8 +429,9 @@ mod tests { .doc_batch .as_ref() .unwrap() - .num_bytes(), - 20 + .doc_buffer + .len(), + 12 ); assert_eq!( ingest_request.subrequests[1] @@ -462,21 +458,4 @@ mod tests { [hola_doc_uid, mundo_doc_uid] ); } - - #[test] - fn test_estimate_size() { - let doc_batch = DocBatchV2 { - doc_buffer: Vec::new().into(), - doc_lengths: Vec::new(), - doc_uids: Vec::new(), - }; - assert_eq!(estimate_size(&doc_batch), ByteSize(0)); - - let doc_batch = DocBatchV2 { - doc_buffer: vec![0u8; 100].into(), - doc_lengths: vec![10, 20, 30], - doc_uids: Vec::new(), - }; - assert_eq!(estimate_size(&doc_batch), ByteSize(118)); - } } diff --git a/quickwit/quickwit-ingest/src/ingest_v2/mrecordlog_utils.rs b/quickwit/quickwit-ingest/src/ingest_v2/mrecordlog_utils.rs index 8a7b039053a..c98bed3eb0f 100644 --- a/quickwit/quickwit-ingest/src/ingest_v2/mrecordlog_utils.rs +++ b/quickwit/quickwit-ingest/src/ingest_v2/mrecordlog_utils.rs @@ -24,8 +24,8 @@ use quickwit_proto::types::{Position, QueueId}; use tracing::instrument; use super::mrecord::MRECORD_HEADER_LEN; +use crate::MRecord; use crate::mrecordlog_async::MultiRecordLogAsync; -use crate::{MRecord, estimate_size}; #[derive(Debug, thiserror::Error)] pub(super) enum AppendDocBatchError { @@ -46,7 +46,7 @@ pub(super) enum AppendDocBatchError { fields( queue_id, num_docs = doc_batch.num_docs(), - num_bytes = doc_batch.num_bytes(), + num_bytes = doc_batch.doc_buffer.len(), force_commit, ) )] @@ -55,7 +55,7 @@ pub(super) async fn append_non_empty_doc_batch( queue_id: &QueueId, doc_batch: DocBatchV2, force_commit: bool, -) -> Result { +) -> Result<(Position, ByteSize), AppendDocBatchError> { let append_result = if force_commit { let encoded_mrecords = doc_batch .into_docs() @@ -87,8 +87,15 @@ pub(super) async fn append_non_empty_doc_batch( .await }; match append_result { - Ok(Some(offset)) => Ok(Position::offset(offset)), - Ok(None) => panic!("`doc_batch` should not be empty"), + Ok(outcome) => { + let offset = outcome + .last_position + .expect("`doc_batch` should not be empty"); + Ok(( + Position::offset(offset), + ByteSize::b(outcome.queue_size_bytes as u64), + )) + } Err(AppendError::IoError(io_error)) => Err(AppendDocBatchError::Io(io_error)), Err(AppendError::MissingQueue(queue_id)) => { Err(AppendDocBatchError::QueueNotFound(queue_id)) @@ -100,20 +107,11 @@ pub(super) async fn append_non_empty_doc_batch( } pub(super) fn doc_batch_size(doc_batch: &DocBatchV2, force_commit: bool) -> ByteSize { - let estimated_size = estimate_size(doc_batch); - if force_commit { - estimated_size + ByteSize::b(MRECORD_HEADER_LEN as u64) - } else { - estimated_size + if doc_batch.is_empty() { + return ByteSize::b(0); } -} - -pub(super) fn read_queue_size(mrecordlog: &MultiRecordLogAsync, queue_id: &QueueId) -> ByteSize { - let num_bytes: usize = mrecordlog - .range(queue_id, ..) - .expect("queue should exist") - .map(|record| record.payload.len()) - .sum(); + let num_records = doc_batch.num_docs() + usize::from(force_commit); + let num_bytes = doc_batch.doc_buffer.len() + num_records * MRECORD_HEADER_LEN; ByteSize::b(num_bytes as u64) } @@ -216,42 +214,68 @@ mod tests { mrecordlog.create_queue(&queue_id).await.unwrap(); - let position = + let (position, queue_size) = append_non_empty_doc_batch(&mut mrecordlog, &queue_id, doc_batch.clone(), false) .await .unwrap(); assert_eq!(position, Position::offset(0u64)); + assert_eq!(queue_size, ByteSize::b(14)); - let position = + let (position, queue_size) = append_non_empty_doc_batch(&mut mrecordlog, &queue_id, doc_batch.clone(), true) .await .unwrap(); assert_eq!(position, Position::offset(2u64)); + assert_eq!(queue_size, ByteSize::b(30)); + } + + #[test] + fn test_empty_doc_batch_size() { + let doc_batch = DocBatchV2::default(); + assert_eq!(doc_batch_size(&doc_batch, false), ByteSize::b(0)); + assert_eq!(doc_batch_size(&doc_batch, true), ByteSize::b(0)); } #[cfg(not(feature = "failpoints"))] #[tokio::test] - async fn test_doc_batch_size_and_read_queue_size() { + async fn test_doc_batch_size_after_repeated_append_and_truncate() { let tempdir = tempfile::tempdir().unwrap(); let mut mrecordlog = MultiRecordLogAsync::open(tempdir.path()).await.unwrap(); let queue_id = "test-queue".to_string(); mrecordlog.create_queue(&queue_id).await.unwrap(); - assert_eq!(read_queue_size(&mrecordlog, &queue_id), ByteSize::b(0)); let doc_batch = DocBatchV2::for_test(["test-doc-foo", "test-doc-bar"]); - assert_eq!(doc_batch_size(&doc_batch, false), ByteSize::b(28)); - assert_eq!(doc_batch_size(&doc_batch, true), ByteSize::b(30)); - - append_non_empty_doc_batch(&mut mrecordlog, &queue_id, doc_batch.clone(), false) - .await - .unwrap(); - assert_eq!(read_queue_size(&mrecordlog, &queue_id), ByteSize::b(28)); - - append_non_empty_doc_batch(&mut mrecordlog, &queue_id, doc_batch, true) + let mut retained_bytes = 0; + let mut next_position = 0u64; + for iteration in 0..32 { + let force_commit = iteration % 2 == 0; + let batch_bytes = if force_commit { 30 } else { 28 }; + assert_eq!( + doc_batch_size(&doc_batch, force_commit), + ByteSize::b(batch_bytes) + ); + + let (_, queue_size) = append_non_empty_doc_batch( + &mut mrecordlog, + &queue_id, + doc_batch.clone(), + force_commit, + ) .await .unwrap(); - assert_eq!(read_queue_size(&mrecordlog, &queue_id), ByteSize::b(58)); + assert_eq!(queue_size.as_u64(), retained_bytes + batch_bytes); + + let outcome = mrecordlog.truncate(&queue_id, next_position).await.unwrap(); + assert_eq!(outcome.evicted_bytes as u64, retained_bytes + 14); + retained_bytes = if force_commit { 16 } else { 14 }; + assert_eq!(outcome.queue_size_bytes as u64, retained_bytes); + assert_eq!( + mrecordlog.summary().queues[&queue_id].num_bytes as u64, + retained_bytes + ); + next_position += 2 + u64::from(force_commit); + } } // This test should be run manually and independently of other tests with the `failpoints` diff --git a/quickwit/quickwit-ingest/src/ingest_v2/state.rs b/quickwit/quickwit-ingest/src/ingest_v2/state.rs index a8745d03102..333839694be 100644 --- a/quickwit/quickwit-ingest/src/ingest_v2/state.rs +++ b/quickwit/quickwit-ingest/src/ingest_v2/state.rs @@ -42,7 +42,6 @@ use tracing::{error, info, instrument}; use super::local_shards::{ShardInfo, ShardInfos, ShardThroughputReadings}; use super::metrics::report_local_shards_metrics; use super::models::IngesterShard; -use super::mrecordlog_utils::read_queue_size; use super::rate_meter::RateMeter; use super::wal_capacity_tracker::WalCapacityTracker; use crate::OpenShardCounts; @@ -393,7 +392,7 @@ impl IngesterState { .checked_sub(1) .map(Position::offset) .unwrap_or(Position::Beginning); - let queue_size = read_queue_size(&mrecordlog, &queue_id); + let queue_size = ByteSize::b(queue_summary.num_bytes as u64); let rate_limiter = RateLimiter::from_settings(rate_limiter_settings); let rate_meter = RateMeter::default(); @@ -717,12 +716,8 @@ impl FullyLockedIngesterState<'_> { .truncate(queue_id, truncate_up_to_offset_inclusive) .await { - Ok(evicted_size) => { - let queue_size = shard - .queue_size - .as_u64() - .saturating_sub(evicted_size.as_u64()); - shard.queue_size = ByteSize::b(queue_size); + Ok(outcome) => { + shard.queue_size = ByteSize::b(outcome.queue_size_bytes as u64); } Err(TruncateError::MissingQueue(_)) => { error!("failed to truncate shard `{queue_id}`: WAL queue not found"); diff --git a/quickwit/quickwit-ingest/src/mrecordlog_async.rs b/quickwit/quickwit-ingest/src/mrecordlog_async.rs index fb83d167f2f..c51f84dc30b 100644 --- a/quickwit/quickwit-ingest/src/mrecordlog_async.rs +++ b/quickwit/quickwit-ingest/src/mrecordlog_async.rs @@ -17,7 +17,6 @@ use std::ops::RangeBounds; use std::path::Path; use bytes::Buf; -use bytesize::ByteSize; use mrecordlog::error::*; use mrecordlog::{MultiRecordLog, PersistAction, PersistPolicy, Record, ResourceUsage}; use tokio::task::JoinError; @@ -137,7 +136,7 @@ impl MultiRecordLogAsync { queue: &str, position_opt: Option, payloads: T, - ) -> Result, AppendError> { + ) -> Result { let span = info_span!("mrecordlog.append_records", queue); let queue = queue.to_string(); self.run_operation(span, move |mrecordlog| { @@ -146,7 +145,6 @@ impl MultiRecordLogAsync { .inspect(|outcome| { WAL_BYTES_WRITTEN_APPEND.inc_by(outcome.wal_bytes_written); }) - .map(|outcome| outcome.last_position) }) .await } @@ -156,16 +154,13 @@ impl MultiRecordLogAsync { &mut self, queue: &str, position: u64, - ) -> Result { + ) -> Result { let span = info_span!("mrecordlog.truncate", queue, position); let queue = queue.to_string(); self.run_operation(span, move |mrecordlog| { - mrecordlog - .truncate(&queue, ..=position) - .inspect(|outcome| { - WAL_BYTES_WRITTEN_TRUNCATE.inc_by(outcome.wal_bytes_written); - }) - .map(|outcome| ByteSize::b(outcome.evicted_bytes as u64)) + mrecordlog.truncate(&queue, ..=position).inspect(|outcome| { + WAL_BYTES_WRITTEN_TRUNCATE.inc_by(outcome.wal_bytes_written); + }) }) .await } diff --git a/quickwit/quickwit-ingest/src/queue.rs b/quickwit/quickwit-ingest/src/queue.rs index 13d87bf33bb..2d1cc684f39 100644 --- a/quickwit/quickwit-ingest/src/queue.rs +++ b/quickwit/quickwit-ingest/src/queue.rs @@ -143,14 +143,14 @@ impl Queues { let real_queue_id = format!("{QUICKWIT_CF_PREFIX}{queue_id}"); // TODO None means we don't have itempotent inserts - let max_position = ctx + let outcome = ctx .protect_future( self.record_log .append_records(&real_queue_id, None, records_it), ) .await?; - Ok(max_position) + Ok(outcome.last_position) } // Streams messages from in `]after_position, +∞[`. diff --git a/quickwit/quickwit-opentelemetry/src/otlp/logs.rs b/quickwit/quickwit-opentelemetry/src/otlp/logs.rs index 27741b03bb7..0f5d53b40f8 100644 --- a/quickwit/quickwit-opentelemetry/src/otlp/logs.rs +++ b/quickwit/quickwit-opentelemetry/src/otlp/logs.rs @@ -240,7 +240,7 @@ impl OtlpGrpcLogsService { if num_log_records == num_parse_errors { return Err(tonic::Status::internal(error_message)); } - let num_bytes = doc_batch.num_bytes() as u64; + let num_bytes = doc_batch.doc_buffer.len() as u64; self.store_logs(index_id.clone(), doc_batch).await?; let labels = @@ -282,7 +282,7 @@ impl OtlpGrpcLogsService { let doc_batch = doc_batch_builder.build(); let current_span = RuntimeSpan::current(); current_span.record("num_log_records", num_log_records); - current_span.record("num_bytes", doc_batch.num_bytes()); + current_span.record("num_bytes", doc_batch.doc_buffer.len()); current_span.record("num_parse_errors", num_parse_errors); let parsed_logs = ParsedLogRecords { @@ -294,7 +294,7 @@ impl OtlpGrpcLogsService { Ok(parsed_logs) } - #[instrument(skip_all, fields(num_bytes = doc_batch.num_bytes()))] + #[instrument(skip_all, fields(num_bytes = doc_batch.doc_buffer.len()))] async fn store_logs( &mut self, index_id: String, diff --git a/quickwit/quickwit-opentelemetry/src/otlp/traces.rs b/quickwit/quickwit-opentelemetry/src/otlp/traces.rs index e9cc8e94096..eafa459ae79 100644 --- a/quickwit/quickwit-opentelemetry/src/otlp/traces.rs +++ b/quickwit/quickwit-opentelemetry/src/otlp/traces.rs @@ -702,7 +702,7 @@ impl OtlpGrpcTracesService { if num_spans == num_parse_errors { return Err(tonic::Status::internal(error_message)); } - let num_bytes = doc_batch.num_bytes() as u64; + let num_bytes = doc_batch.doc_buffer.len() as u64; self.store_spans(index_id.clone(), doc_batch).await?; let labels = @@ -744,7 +744,7 @@ impl OtlpGrpcTracesService { let doc_batch = doc_batch_builder.build(); let current_span = RuntimeSpan::current(); current_span.record("num_spans", num_spans); - current_span.record("num_bytes", doc_batch.num_bytes()); + current_span.record("num_bytes", doc_batch.doc_buffer.len()); current_span.record("num_parse_errors", num_parse_errors); let parsed_spans = ParsedSpans { @@ -756,7 +756,7 @@ impl OtlpGrpcTracesService { Ok(parsed_spans) } - #[instrument(skip_all, fields(num_bytes = doc_batch.num_bytes()))] + #[instrument(skip_all, fields(num_bytes = doc_batch.doc_buffer.len()))] async fn store_spans( &mut self, index_id: String, diff --git a/quickwit/quickwit-proto/src/ingest/ingester.rs b/quickwit/quickwit-proto/src/ingest/ingester.rs index 46d64a9faf7..fce3650524c 100644 --- a/quickwit/quickwit-proto/src/ingest/ingester.rs +++ b/quickwit/quickwit-proto/src/ingest/ingester.rs @@ -57,9 +57,9 @@ impl FetchPayload { } } - pub fn estimate_size(&self) -> ByteSize { + pub fn buffer_size(&self) -> ByteSize { if let Some(mrecord_batch) = &self.mrecord_batch { - mrecord_batch.estimate_size() + mrecord_batch.buffer_size() } else { ByteSize(0) } diff --git a/quickwit/quickwit-proto/src/ingest/mod.rs b/quickwit/quickwit-proto/src/ingest/mod.rs index 9e6f81a26af..57550e9aabb 100644 --- a/quickwit/quickwit-proto/src/ingest/mod.rs +++ b/quickwit/quickwit-proto/src/ingest/mod.rs @@ -164,10 +164,6 @@ impl DocBatchV2 { self.doc_lengths.is_empty() } - pub fn num_bytes(&self) -> usize { - self.doc_buffer.len() + self.doc_lengths.len() * 4 - } - pub fn num_docs(&self) -> usize { self.doc_lengths.len() } @@ -207,7 +203,7 @@ impl MRecordBatch { self.mrecord_lengths.is_empty() } - pub fn estimate_size(&self) -> ByteSize { + pub fn buffer_size(&self) -> ByteSize { ByteSize((self.mrecord_buffer.len() + self.mrecord_lengths.len() * 4) as u64) } diff --git a/quickwit/quickwit-serve/src/elasticsearch_api/bulk_v2.rs b/quickwit/quickwit-serve/src/elasticsearch_api/bulk_v2.rs index 18159866f41..b03b48aba89 100644 --- a/quickwit/quickwit-serve/src/elasticsearch_api/bulk_v2.rs +++ b/quickwit/quickwit-serve/src/elasticsearch_api/bulk_v2.rs @@ -458,13 +458,19 @@ mod tests { assert_eq!(subrequests[0].index_id, "my-index-1"); assert_eq!(subrequests[0].source_id, INGEST_V2_SOURCE_ID); assert_eq!(subrequests[0].doc_batch.as_ref().unwrap().num_docs(), 2); - assert_eq!(subrequests[0].doc_batch.as_ref().unwrap().num_bytes(), 104); + assert_eq!( + subrequests[0].doc_batch.as_ref().unwrap().doc_buffer.len(), + 96 + ); assert_eq!(subrequests[1].subrequest_id, 1); assert_eq!(subrequests[1].index_id, "my-index-2"); assert_eq!(subrequests[1].source_id, INGEST_V2_SOURCE_ID); assert_eq!(subrequests[1].doc_batch.as_ref().unwrap().num_docs(), 1); - assert_eq!(subrequests[1].doc_batch.as_ref().unwrap().num_bytes(), 52); + assert_eq!( + subrequests[1].doc_batch.as_ref().unwrap().doc_buffer.len(), + 48 + ); Ok(IngestResponseV2 { successes: vec![ @@ -572,7 +578,10 @@ mod tests { assert_eq!(subrequest_0.index_id, "my-index-1"); assert_eq!(subrequest_0.source_id, INGEST_V2_SOURCE_ID); assert_eq!(subrequest_0.doc_batch.as_ref().unwrap().num_docs(), 1); - assert_eq!(subrequest_0.doc_batch.as_ref().unwrap().num_bytes(), 52); + assert_eq!( + subrequest_0.doc_batch.as_ref().unwrap().doc_buffer.len(), + 48 + ); Ok(IngestResponseV2 { successes: vec![IngestSuccess {