From f30302fdc69d6999db552f041f90b1e702e4b132 Mon Sep 17 00:00:00 2001 From: "francois.massot" Date: Fri, 2 Oct 2026 23:09:39 +0200 Subject: [PATCH 1/7] Optimize native ingest for searchbench --- quickwit/quickwit-common/src/shared_consts.rs | 6 +- .../quickwit-ingest/benches/mrecord_bench.rs | 44 ++++++++++- quickwit/quickwit-ingest/src/ingest_v2/mod.rs | 9 +++ .../src/ingest_api/rest_handler.rs | 79 ++++++++++++++++--- 4 files changed, 122 insertions(+), 16 deletions(-) diff --git a/quickwit/quickwit-common/src/shared_consts.rs b/quickwit/quickwit-common/src/shared_consts.rs index 5cb5601fe4c..48c61df3156 100644 --- a/quickwit/quickwit-common/src/shared_consts.rs +++ b/quickwit/quickwit-common/src/shared_consts.rs @@ -85,8 +85,10 @@ pub const DEFAULT_SHARD_THROUGHPUT_LIMIT: ByteSize = ByteSize::mib(5); /// Large enough to absorb small bursts but should remain defensive against unbalanced shards. pub const DEFAULT_SHARD_BURST_LIMIT: ByteSize = ByteSize::mib(50); -/// A compromise between "exponential" scale up and moderate shard count increase. -pub const DEFAULT_SHARD_SCALE_UP_FACTOR: f32 = 1.5; +/// Scale ingest shards aggressively on sustained bursts. Starting from one shard, a factor of 4 +/// reaches 16 shards in two scaling rounds instead of six with a factor of 1.5, which avoids long +/// rate-limited warm-up phases on high-throughput ingest workloads. +pub const DEFAULT_SHARD_SCALE_UP_FACTOR: f32 = 4.0; // (Just a reexport). pub use bytesize::MIB; diff --git a/quickwit/quickwit-ingest/benches/mrecord_bench.rs b/quickwit/quickwit-ingest/benches/mrecord_bench.rs index 228a5567877..5e750d3d914 100644 --- a/quickwit/quickwit-ingest/benches/mrecord_bench.rs +++ b/quickwit/quickwit-ingest/benches/mrecord_bench.rs @@ -28,7 +28,8 @@ use std::time::{Duration, Instant}; use bytes::{BufMut, Bytes}; use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; use mrecordlog::MultiRecordLog; -use quickwit_ingest::MRecord; +use quickwit_ingest::{DocBatchV2Builder, MRecord}; +use quickwit_proto::types::DocUidGenerator; /// Representative document sizes (bytes): a small log line up to a large structured event. const DOC_SIZES: [usize; 4] = [128, 1_024, 8_192, 65_536]; @@ -180,5 +181,44 @@ fn bench_append_batch(criterion: &mut Criterion) { group.finish(); } -criterion_group!(benches, bench_encode, bench_decode, bench_append_batch); +fn bench_doc_batch_builder(criterion: &mut Criterion) { + let mut group = criterion.benchmark_group("doc_batch_v2_builder"); + let docs: Vec = (0..10_000) + .map(|index| make_doc(BATCH_DOC_SIZES[index % BATCH_DOC_SIZES.len()])) + .collect(); + let total_bytes: usize = docs.iter().map(Bytes::len).sum(); + group.throughput(Throughput::Bytes(total_bytes as u64)); + + group.bench_function("default", |bencher| { + bencher.iter(|| { + let mut builder = DocBatchV2Builder::default(); + let mut doc_uid_generator = DocUidGenerator::default(); + for doc in &docs { + builder.add_doc(doc_uid_generator.next_doc_uid(), doc); + } + black_box(builder.build()) + }); + }); + + group.bench_function("preallocated", |bencher| { + bencher.iter(|| { + let mut builder = DocBatchV2Builder::with_capacity(total_bytes); + let mut doc_uid_generator = DocUidGenerator::default(); + for doc in &docs { + builder.add_doc(doc_uid_generator.next_doc_uid(), doc); + } + black_box(builder.build()) + }); + }); + + group.finish(); +} + +criterion_group!( + benches, + bench_encode, + bench_decode, + bench_append_batch, + bench_doc_batch_builder +); criterion_main!(benches); diff --git a/quickwit/quickwit-ingest/src/ingest_v2/mod.rs b/quickwit/quickwit-ingest/src/ingest_v2/mod.rs index 5a6c6e2c791..e77a3dc358d 100644 --- a/quickwit/quickwit-ingest/src/ingest_v2/mod.rs +++ b/quickwit/quickwit-ingest/src/ingest_v2/mod.rs @@ -153,6 +153,15 @@ pub struct DocBatchV2Builder { } impl DocBatchV2Builder { + /// Creates a builder with enough room for `doc_buffer_capacity` bytes of document payloads. + pub fn with_capacity(doc_buffer_capacity: usize) -> Self { + Self { + doc_uids: Vec::new(), + doc_buffer: BytesMut::with_capacity(doc_buffer_capacity), + doc_lengths: Vec::new(), + } + } + /// Adds a document to the batch. pub fn add_doc(&mut self, doc_uid: DocUid, doc: &[u8]) { self.doc_uids.push(doc_uid); diff --git a/quickwit/quickwit-serve/src/ingest_api/rest_handler.rs b/quickwit/quickwit-serve/src/ingest_api/rest_handler.rs index 9040c8fd700..d3e48d447cd 100644 --- a/quickwit/quickwit-serve/src/ingest_api/rest_handler.rs +++ b/quickwit/quickwit-serve/src/ingest_api/rest_handler.rs @@ -15,13 +15,13 @@ use bytes::{Buf, Bytes}; use quickwit_config::{INGEST_V2_SOURCE_ID, IngestApiConfig, validate_identifier}; use quickwit_ingest::{ - CommitType, DocBatchBuilder, DocBatchV2Builder, FetchResponse, IngestRequest, IngestService, - IngestServiceClient, IngestServiceError, TailRequest, + CommitType, DocBatchBuilder, FetchResponse, IngestRequest, IngestService, IngestServiceClient, + IngestServiceError, TailRequest, }; -use quickwit_proto::ingest::CommitTypeV2; use quickwit_proto::ingest::router::{ IngestRequestV2, IngestRouterService, IngestRouterServiceClient, IngestSubrequest, }; +use quickwit_proto::ingest::{CommitTypeV2, DocBatchV2}; use quickwit_proto::types::{DocUidGenerator, IndexId}; use serde::Deserialize; use warp::{Filter, Rejection}; @@ -193,14 +193,7 @@ async fn ingest_v2( ingest_options: IngestOptions, ingest_router: IngestRouterServiceClient, ) -> Result { - let mut doc_batch_builder = DocBatchV2Builder::default(); - let mut doc_uid_generator = DocUidGenerator::default(); - - for doc in lines(&body.content) { - doc_batch_builder.add_doc(doc_uid_generator.next_doc_uid(), doc); - } - drop(body); - let doc_batch_opt = doc_batch_builder.build(); + let doc_batch_opt = build_doc_batch_v2_from_ndjson_body(body.content); let Some(doc_batch) = doc_batch_opt else { let response = RestIngestResponse::default(); @@ -279,6 +272,51 @@ pub(crate) fn lines(body: &Bytes) -> impl Iterator { .filter(|line| !is_empty_or_blank_line(line)) } +fn build_doc_batch_v2_from_ndjson_body(doc_buffer: Bytes) -> Option { + let mut doc_uids = Vec::new(); + let mut doc_lengths = Vec::new(); + let mut doc_uid_generator = DocUidGenerator::default(); + let mut segment_start = 0usize; + let mut line_start = 0usize; + + for (position, byte) in doc_buffer.iter().enumerate() { + if *byte != b'\n' { + continue; + } + let line = &doc_buffer[line_start..position]; + if !is_empty_or_blank_line(line) { + doc_uids.push(doc_uid_generator.next_doc_uid()); + doc_lengths.push((position + 1 - segment_start) as u32); + segment_start = position + 1; + } + line_start = position + 1; + } + + let line = &doc_buffer[line_start..]; + if !is_empty_or_blank_line(line) { + doc_uids.push(doc_uid_generator.next_doc_uid()); + doc_lengths.push((doc_buffer.len() - segment_start) as u32); + segment_start = doc_buffer.len(); + } + + if doc_uids.is_empty() { + return None; + } + if segment_start < doc_buffer.len() { + let trailing_whitespace_len = doc_buffer.len() - segment_start; + let last_doc_len = doc_lengths + .last_mut() + .expect("doc lengths should not be empty"); + *last_doc_len += trailing_whitespace_len as u32; + } + + Some(DocBatchV2 { + doc_uids, + doc_buffer, + doc_lengths, + }) +} + #[inline] fn is_empty_or_blank_line(line: &[u8]) -> bool { line.is_empty() || line.iter().all(|ch| ch.is_ascii_whitespace()) @@ -298,7 +336,7 @@ pub(crate) mod tests { }; use quickwit_proto::ingest::router::IngestRouterServiceClient; - use super::{RestIngestResponse, ingest_api_handlers}; + use super::{RestIngestResponse, build_doc_batch_v2_from_ndjson_body, ingest_api_handlers}; use crate::ingest_api::lines; #[test] @@ -319,6 +357,23 @@ pub(crate) mod tests { } } + #[test] + fn test_build_doc_batch_v2_from_ndjson_body_zero_copy() { + let body = Bytes::from_static(b"\n {\"id\":1}\n\n{\"id\":2}\n \n"); + let doc_batch = build_doc_batch_v2_from_ndjson_body(body.clone()).unwrap(); + assert_eq!(doc_batch.num_docs(), 2); + assert_eq!(doc_batch.doc_buffer, body); + + let docs: Vec = doc_batch.docs().map(|(_doc_uid, doc)| doc).collect(); + assert_eq!(str::from_utf8(&docs[0]).unwrap(), "\n {\"id\":1}\n"); + assert_eq!(str::from_utf8(&docs[1]).unwrap(), "\n{\"id\":2}\n \n"); + } + + #[test] + fn test_build_doc_batch_v2_from_blank_body() { + assert!(build_doc_batch_v2_from_ndjson_body(Bytes::from_static(b"\n \n\t")).is_none()); + } + pub(crate) async fn setup_ingest_v1_service( queues: &[&str], config: &IngestApiConfig, From 527b4d4e7a82ca467fbfa5bfa6dece7ac8d7ccb4 Mon Sep 17 00:00:00 2001 From: "francois.massot" Date: Fri, 2 Oct 2026 23:24:06 +0200 Subject: [PATCH 2/7] Keep default ingest shard scale up factor --- quickwit/quickwit-common/src/shared_consts.rs | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/quickwit/quickwit-common/src/shared_consts.rs b/quickwit/quickwit-common/src/shared_consts.rs index 48c61df3156..5cb5601fe4c 100644 --- a/quickwit/quickwit-common/src/shared_consts.rs +++ b/quickwit/quickwit-common/src/shared_consts.rs @@ -85,10 +85,8 @@ pub const DEFAULT_SHARD_THROUGHPUT_LIMIT: ByteSize = ByteSize::mib(5); /// Large enough to absorb small bursts but should remain defensive against unbalanced shards. pub const DEFAULT_SHARD_BURST_LIMIT: ByteSize = ByteSize::mib(50); -/// Scale ingest shards aggressively on sustained bursts. Starting from one shard, a factor of 4 -/// reaches 16 shards in two scaling rounds instead of six with a factor of 1.5, which avoids long -/// rate-limited warm-up phases on high-throughput ingest workloads. -pub const DEFAULT_SHARD_SCALE_UP_FACTOR: f32 = 4.0; +/// A compromise between "exponential" scale up and moderate shard count increase. +pub const DEFAULT_SHARD_SCALE_UP_FACTOR: f32 = 1.5; // (Just a reexport). pub use bytesize::MIB; From c5478efb985bbef204b9d39755aaa000a3087c84 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fran=C3=A7ois=20Massot?= Date: Sat, 10 Oct 2026 18:59:34 +0200 Subject: [PATCH 3/7] Find NDJSON newlines with memchr build_doc_batch_v2_from_ndjson_body scanned the body byte by byte. memchr finds the newlines with SIMD. On 200 real NDJSON bodies (2,000 OTel log docs each, 2.3 MB, ~1.2 KB per doc), same output on every body: - byte by byte: 3.5 GB/s, 670 us per body - memchr: 40.7 GB/s, 58 us per body (x11.6, ~0.3 us saved per doc) At ~1.1M docs/s that is ~0.34 cores saved, and each request holds its tokio worker ~0.06 ms instead of ~0.67 ms. --- quickwit/Cargo.lock | 1 + quickwit/Cargo.toml | 1 + quickwit/quickwit-serve/Cargo.toml | 1 + quickwit/quickwit-serve/src/ingest_api/rest_handler.rs | 6 ++---- 4 files changed, 5 insertions(+), 4 deletions(-) diff --git a/quickwit/Cargo.lock b/quickwit/Cargo.lock index 4c929de6fc0..fc9b7393eed 100644 --- a/quickwit/Cargo.lock +++ b/quickwit/Cargo.lock @@ -8008,6 +8008,7 @@ dependencies = [ "humantime", "hyper-util", "itertools 0.15.0", + "memchr", "mime_guess", "mockall", "opentelemetry", diff --git a/quickwit/Cargo.toml b/quickwit/Cargo.toml index 1ade0c4c0e2..a58d5f3797f 100644 --- a/quickwit/Cargo.toml +++ b/quickwit/Cargo.toml @@ -126,6 +126,7 @@ libz-sys = "1.1" lru = "0.18" matches = "0.1" md5 = "0.8" +memchr = "2" metrics = "0.24" metrics-opentelemetry = "=0.24032.8" metrics-exporter-prometheus = { version = "0.18", default-features = false } diff --git a/quickwit/quickwit-serve/Cargo.toml b/quickwit/quickwit-serve/Cargo.toml index 0d554ffc2ee..3f4724cbe9e 100644 --- a/quickwit/quickwit-serve/Cargo.toml +++ b/quickwit/quickwit-serve/Cargo.toml @@ -28,6 +28,7 @@ http-serde = { workspace = true } humantime = { workspace = true } hyper-util = { workspace = true } itertools = { workspace = true } +memchr = { workspace = true } mime_guess = { workspace = true } opentelemetry = { workspace = true } percent-encoding = { workspace = true } diff --git a/quickwit/quickwit-serve/src/ingest_api/rest_handler.rs b/quickwit/quickwit-serve/src/ingest_api/rest_handler.rs index d3e48d447cd..50813a6ea27 100644 --- a/quickwit/quickwit-serve/src/ingest_api/rest_handler.rs +++ b/quickwit/quickwit-serve/src/ingest_api/rest_handler.rs @@ -279,10 +279,8 @@ fn build_doc_batch_v2_from_ndjson_body(doc_buffer: Bytes) -> Option let mut segment_start = 0usize; let mut line_start = 0usize; - for (position, byte) in doc_buffer.iter().enumerate() { - if *byte != b'\n' { - continue; - } + // memchr searches for newlines with SIMD: ~10x faster than a byte-by-byte scan. + for position in memchr::memchr_iter(b'\n', &doc_buffer) { let line = &doc_buffer[line_start..position]; if !is_empty_or_blank_line(line) { doc_uids.push(doc_uid_generator.next_doc_uid()); From deaba3077dda1f4949caed8fafc877de1e9cfdc3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fran=C3=A7ois=20Massot?= Date: Sat, 10 Oct 2026 19:07:27 +0200 Subject: [PATCH 4/7] Rename doc_buffer to ndjson_body in build_doc_batch_v2_from_ndjson_body --- .../src/ingest_api/rest_handler.rs | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/quickwit/quickwit-serve/src/ingest_api/rest_handler.rs b/quickwit/quickwit-serve/src/ingest_api/rest_handler.rs index 50813a6ea27..15db1bf9796 100644 --- a/quickwit/quickwit-serve/src/ingest_api/rest_handler.rs +++ b/quickwit/quickwit-serve/src/ingest_api/rest_handler.rs @@ -272,7 +272,7 @@ pub(crate) fn lines(body: &Bytes) -> impl Iterator { .filter(|line| !is_empty_or_blank_line(line)) } -fn build_doc_batch_v2_from_ndjson_body(doc_buffer: Bytes) -> Option { +fn build_doc_batch_v2_from_ndjson_body(ndjson_body: Bytes) -> Option { let mut doc_uids = Vec::new(); let mut doc_lengths = Vec::new(); let mut doc_uid_generator = DocUidGenerator::default(); @@ -280,8 +280,8 @@ fn build_doc_batch_v2_from_ndjson_body(doc_buffer: Bytes) -> Option let mut line_start = 0usize; // memchr searches for newlines with SIMD: ~10x faster than a byte-by-byte scan. - for position in memchr::memchr_iter(b'\n', &doc_buffer) { - let line = &doc_buffer[line_start..position]; + for position in memchr::memchr_iter(b'\n', &ndjson_body) { + let line = &ndjson_body[line_start..position]; if !is_empty_or_blank_line(line) { doc_uids.push(doc_uid_generator.next_doc_uid()); doc_lengths.push((position + 1 - segment_start) as u32); @@ -290,18 +290,18 @@ fn build_doc_batch_v2_from_ndjson_body(doc_buffer: Bytes) -> Option line_start = position + 1; } - let line = &doc_buffer[line_start..]; + let line = &ndjson_body[line_start..]; if !is_empty_or_blank_line(line) { doc_uids.push(doc_uid_generator.next_doc_uid()); - doc_lengths.push((doc_buffer.len() - segment_start) as u32); - segment_start = doc_buffer.len(); + doc_lengths.push((ndjson_body.len() - segment_start) as u32); + segment_start = ndjson_body.len(); } if doc_uids.is_empty() { return None; } - if segment_start < doc_buffer.len() { - let trailing_whitespace_len = doc_buffer.len() - segment_start; + if segment_start < ndjson_body.len() { + let trailing_whitespace_len = ndjson_body.len() - segment_start; let last_doc_len = doc_lengths .last_mut() .expect("doc lengths should not be empty"); @@ -310,7 +310,7 @@ fn build_doc_batch_v2_from_ndjson_body(doc_buffer: Bytes) -> Option Some(DocBatchV2 { doc_uids, - doc_buffer, + doc_buffer: ndjson_body, doc_lengths, }) } From cc117df0dfad17c4af4851a6ed97b329ac278f30 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fran=C3=A7ois=20Massot?= Date: Sat, 10 Oct 2026 19:20:45 +0200 Subject: [PATCH 5/7] Move NDJSON splitting to quickwit-ingest for reuse Adds quickwit_ingest::{split_ndjson_lines, doc_batch_v2_from_ndjson}: - split_ndjson_lines: iterator over the non-blank lines of an NDJSON body, newlines found with memchr. Replaces quickwit-serve's lines(), so the ingest v1 and Elasticsearch bulk handlers get the memchr speedup too. - doc_batch_v2_from_ndjson: the zero-copy DocBatchV2 builder from the ingest v2 REST handler, moved as is. Tests moved along, plus no-trailing-newline, empty body and a check that both helpers agree. --- quickwit/Cargo.lock | 2 +- quickwit/quickwit-ingest/Cargo.toml | 1 + quickwit/quickwit-ingest/src/lib.rs | 2 + quickwit/quickwit-ingest/src/ndjson.rs | 163 ++++++++++++++++++ quickwit/quickwit-serve/Cargo.toml | 1 - .../src/elasticsearch_api/bulk.rs | 4 +- .../src/elasticsearch_api/bulk_v2.rs | 5 +- quickwit/quickwit-serve/src/ingest_api/mod.rs | 2 +- .../src/ingest_api/rest_handler.rs | 105 +---------- 9 files changed, 179 insertions(+), 106 deletions(-) create mode 100644 quickwit/quickwit-ingest/src/ndjson.rs diff --git a/quickwit/Cargo.lock b/quickwit/Cargo.lock index fc9b7393eed..601594bdcda 100644 --- a/quickwit/Cargo.lock +++ b/quickwit/Cargo.lock @@ -7562,6 +7562,7 @@ dependencies = [ "futures", "http 1.5.0", "itertools 0.15.0", + "memchr", "mockall", "mrecordlog", "prost 0.14.4", @@ -8008,7 +8009,6 @@ dependencies = [ "humantime", "hyper-util", "itertools 0.15.0", - "memchr", "mime_guess", "mockall", "opentelemetry", diff --git a/quickwit/quickwit-ingest/Cargo.toml b/quickwit/quickwit-ingest/Cargo.toml index 4ecaeddf401..ed8ce9f496a 100644 --- a/quickwit/quickwit-ingest/Cargo.toml +++ b/quickwit/quickwit-ingest/Cargo.toml @@ -19,6 +19,7 @@ fail = { workspace = true, optional = true } futures = { workspace = true } http = { workspace = true } itertools = { workspace = true } +memchr = { workspace = true } mockall = { workspace = true, optional = true } mrecordlog = { workspace = true } prost = { workspace = true } diff --git a/quickwit/quickwit-ingest/src/lib.rs b/quickwit/quickwit-ingest/src/lib.rs index 7529bed1ebc..f0c6d36d859 100644 --- a/quickwit/quickwit-ingest/src/lib.rs +++ b/quickwit/quickwit-ingest/src/lib.rs @@ -24,6 +24,7 @@ mod ingest_v2; mod memory_capacity; mod metrics; mod mrecordlog_async; +mod ndjson; mod notifications; mod position; mod queue; @@ -39,6 +40,7 @@ pub use ingest_api_service::{GetMemoryCapacity, GetPartitionId, IngestApiService pub use ingest_service::*; pub use ingest_v2::*; pub use memory_capacity::MemoryCapacity; +pub use ndjson::{doc_batch_v2_from_ndjson, split_ndjson_lines}; pub use position::Position; pub use queue::Queues; use quickwit_actors::{Mailbox, Universe}; diff --git a/quickwit/quickwit-ingest/src/ndjson.rs b/quickwit/quickwit-ingest/src/ndjson.rs new file mode 100644 index 00000000000..53b2625425a --- /dev/null +++ b/quickwit/quickwit-ingest/src/ndjson.rs @@ -0,0 +1,163 @@ +// Copyright 2021-Present Datadog, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Helpers to split NDJSON request bodies into documents. +//! +//! Newlines are found with `memchr` (SIMD), which is ~10x faster than a byte-by-byte scan on +//! typical log payloads. Empty and blank (whitespace-only) lines are skipped. + +use bytes::Bytes; +use quickwit_proto::ingest::DocBatchV2; +use quickwit_proto::types::DocUidGenerator; + +/// Returns an iterator over the non-blank lines of an NDJSON body, without their trailing `\n`. +pub fn split_ndjson_lines(ndjson_body: &[u8]) -> impl Iterator { + let mut line_start = 0; + memchr::memchr_iter(b'\n', ndjson_body) + .chain(std::iter::once(ndjson_body.len())) + .map(move |line_end| { + let line = &ndjson_body[line_start..line_end]; + line_start = line_end + 1; + line + }) + .filter(|line| !is_empty_or_blank_line(line)) +} + +/// Builds a [`DocBatchV2`] from an NDJSON body without copying the documents: the body becomes +/// the batch's doc buffer, and each non-blank line becomes a document with a fresh doc UID. +/// +/// Since the docs must cover the whole buffer, the newlines and the blank lines are kept with the +/// neighboring docs: a doc spans from the end of the previous doc to the end of its own line, +/// including the `\n`, and the last doc also covers any trailing blank lines. JSON parsers ignore +/// this leading and trailing whitespace. +/// +/// Returns `None` if the body has no non-blank line. +pub fn doc_batch_v2_from_ndjson(ndjson_body: Bytes) -> Option { + let mut doc_uids = Vec::new(); + let mut doc_lengths = Vec::new(); + let mut doc_uid_generator = DocUidGenerator::default(); + let mut segment_start = 0usize; + let mut line_start = 0usize; + + for position in memchr::memchr_iter(b'\n', &ndjson_body) { + let line = &ndjson_body[line_start..position]; + if !is_empty_or_blank_line(line) { + doc_uids.push(doc_uid_generator.next_doc_uid()); + doc_lengths.push((position + 1 - segment_start) as u32); + segment_start = position + 1; + } + line_start = position + 1; + } + + let line = &ndjson_body[line_start..]; + if !is_empty_or_blank_line(line) { + doc_uids.push(doc_uid_generator.next_doc_uid()); + doc_lengths.push((ndjson_body.len() - segment_start) as u32); + segment_start = ndjson_body.len(); + } + + if doc_uids.is_empty() { + return None; + } + if segment_start < ndjson_body.len() { + let trailing_whitespace_len = ndjson_body.len() - segment_start; + let last_doc_len = doc_lengths + .last_mut() + .expect("doc lengths should not be empty"); + *last_doc_len += trailing_whitespace_len as u32; + } + + Some(DocBatchV2 { + doc_uids, + doc_buffer: ndjson_body, + doc_lengths, + }) +} + +#[inline] +fn is_empty_or_blank_line(line: &[u8]) -> bool { + line.is_empty() || line.iter().all(|ch| ch.is_ascii_whitespace()) +} + +#[cfg(test)] +mod tests { + use std::str; + + use super::*; + + #[test] + fn test_split_ndjson_lines() { + let test_cases = [ + // an empty line is inserted before the metadata action and the doc + (&b"\n{ \"create\" : { \"_index\" : \"my-index-1\", \"_id\" : \"1\"} }\n{\"id\": 1, \"message\": \"push\"}"[..], 2), + // a blank line is inserted before the metadata action and the doc + (&b" \n{ \"create\" : { \"_index\" : \"my-index-1\", \"_id\" : \"1\"} }\n{\"id\": 1, \"message\": \"push\"}"[..], 2), + // an empty line is inserted after the metadata action and before the doc + (&b"{ \"create\" : { \"_index\" : \"my-index-1\", \"_id\" : \"1\"} }\n\n{\"id\": 1, \"message\": \"push\"}"[..], 2), + // a blank line is inserted after the metadata action and before the doc + (&b"{ \"create\" : { \"_index\" : \"my-index-1\", \"_id\" : \"1\"} }\n \n{\"id\": 1, \"message\": \"push\"}"[..], 2), + ]; + for &(input, expected_count) in &test_cases { + assert_eq!(split_ndjson_lines(input).count(), expected_count); + } + let lines: Vec<&[u8]> = split_ndjson_lines(b"a\n\nbc\n \nd").collect(); + assert_eq!(lines, [&b"a"[..], b"bc", b"d"]); + + let lines: Vec<&[u8]> = split_ndjson_lines(b"a\n").collect(); + assert_eq!(lines, [&b"a"[..]]); + + assert_eq!(split_ndjson_lines(b"").count(), 0); + assert_eq!(split_ndjson_lines(b"\n \n\t").count(), 0); + } + + #[test] + fn test_doc_batch_v2_from_ndjson_zero_copy() { + let body = Bytes::from_static(b"\n {\"id\":1}\n\n{\"id\":2}\n \n"); + let doc_batch = doc_batch_v2_from_ndjson(body.clone()).unwrap(); + assert_eq!(doc_batch.num_docs(), 2); + assert_eq!(doc_batch.doc_buffer, body); + + let docs: Vec = doc_batch.docs().map(|(_doc_uid, doc)| doc).collect(); + assert_eq!(str::from_utf8(&docs[0]).unwrap(), "\n {\"id\":1}\n"); + assert_eq!(str::from_utf8(&docs[1]).unwrap(), "\n{\"id\":2}\n \n"); + } + + #[test] + fn test_doc_batch_v2_from_ndjson_no_trailing_newline() { + let body = Bytes::from_static(b"{\"id\":1}\n{\"id\":2}"); + let doc_batch = doc_batch_v2_from_ndjson(body).unwrap(); + let docs: Vec = doc_batch.docs().map(|(_doc_uid, doc)| doc).collect(); + assert_eq!(docs, [&b"{\"id\":1}\n"[..], b"{\"id\":2}"]); + } + + #[test] + fn test_doc_batch_v2_from_blank_ndjson() { + assert!(doc_batch_v2_from_ndjson(Bytes::new()).is_none()); + assert!(doc_batch_v2_from_ndjson(Bytes::from_static(b"\n \n\t")).is_none()); + } + + #[test] + fn test_doc_batch_v2_from_ndjson_matches_split_ndjson_lines() { + let body = Bytes::from_static(b" \n{\"a\":1}\n\n \t\n{\"b\":2}\n{\"c\":3}\n\n"); + let doc_batch = doc_batch_v2_from_ndjson(body.clone()).unwrap(); + let docs: Vec = doc_batch.docs().map(|(_doc_uid, doc)| doc).collect(); + let lines: Vec<&[u8]> = split_ndjson_lines(&body).collect(); + assert_eq!(docs.len(), lines.len()); + for (doc, line) in docs.iter().zip(&lines) { + assert_eq!(doc.trim_ascii(), *line); + } + let doc_uids: Vec<_> = doc_batch.docs().map(|(doc_uid, _doc)| doc_uid).collect(); + assert!(doc_uids.windows(2).all(|pair| pair[0] < pair[1])); + } +} diff --git a/quickwit/quickwit-serve/Cargo.toml b/quickwit/quickwit-serve/Cargo.toml index 3f4724cbe9e..0d554ffc2ee 100644 --- a/quickwit/quickwit-serve/Cargo.toml +++ b/quickwit/quickwit-serve/Cargo.toml @@ -28,7 +28,6 @@ http-serde = { workspace = true } humantime = { workspace = true } hyper-util = { workspace = true } itertools = { workspace = true } -memchr = { workspace = true } mime_guess = { workspace = true } opentelemetry = { workspace = true } percent-encoding = { workspace = true } diff --git a/quickwit/quickwit-serve/src/elasticsearch_api/bulk.rs b/quickwit/quickwit-serve/src/elasticsearch_api/bulk.rs index c39e3072355..a7574c7a369 100644 --- a/quickwit/quickwit-serve/src/elasticsearch_api/bulk.rs +++ b/quickwit/quickwit-serve/src/elasticsearch_api/bulk.rs @@ -18,6 +18,7 @@ use std::time::Instant; use bytesize::ByteSize; use quickwit_ingest::{ CommitType, DocBatchBuilder, IngestRequest, IngestService, IngestServiceClient, + split_ndjson_lines, }; use quickwit_proto::ingest::router::IngestRouterServiceClient; use quickwit_proto::types::IndexId; @@ -29,7 +30,6 @@ use crate::elasticsearch_api::filter::{elastic_bulk_filter, elastic_index_bulk_f use crate::elasticsearch_api::make_elastic_api_response; use crate::elasticsearch_api::model::{BulkAction, ElasticBulkOptions, ElasticsearchError}; use crate::format::extract_format_from_qs; -use crate::ingest_api::lines; use crate::rest::recover_fn; use crate::{Body, with_arg}; @@ -111,7 +111,7 @@ async fn elastic_ingest_bulk( } let now = Instant::now(); let mut doc_batch_builders = HashMap::new(); - let mut lines = lines(&body.content).enumerate(); + let mut lines = split_ndjson_lines(&body.content).enumerate(); while let Some((line_number, line)) = lines.next() { let action = serde_json::from_slice::(line).map_err(|error| { diff --git a/quickwit/quickwit-serve/src/elasticsearch_api/bulk_v2.rs b/quickwit/quickwit-serve/src/elasticsearch_api/bulk_v2.rs index 18159866f41..a101cfea321 100644 --- a/quickwit/quickwit-serve/src/elasticsearch_api/bulk_v2.rs +++ b/quickwit/quickwit-serve/src/elasticsearch_api/bulk_v2.rs @@ -17,7 +17,7 @@ use std::time::Instant; use quickwit_common::rate_limited_error; use quickwit_config::{INGEST_V2_SOURCE_ID, validate_identifier}; -use quickwit_ingest::IngestRequestV2Builder; +use quickwit_ingest::{IngestRequestV2Builder, split_ndjson_lines}; use quickwit_proto::ingest::CommitTypeV2; use quickwit_proto::ingest::router::{ IngestFailureReason, IngestResponseV2, IngestRouterService, IngestRouterServiceClient, @@ -29,7 +29,6 @@ use warp::hyper::StatusCode; use super::model::ElasticException; use crate::Body; use crate::elasticsearch_api::model::{BulkAction, ElasticBulkOptions, ElasticsearchError}; -use crate::ingest_api::lines; #[derive(Debug, Default, Serialize, Deserialize)] pub(crate) struct ElasticBulkResponse { @@ -88,7 +87,7 @@ pub(crate) async fn elastic_bulk_ingest_v2( ) -> Result { let now = Instant::now(); let mut ingest_request_builder = IngestRequestV2Builder::default(); - let mut lines = lines(&body.content).enumerate(); + let mut lines = split_ndjson_lines(&body.content).enumerate(); let mut per_subrequest_doc_handles: HashMap> = HashMap::new(); let mut action_count = 0; let mut invalid_index_id_items = Vec::new(); diff --git a/quickwit/quickwit-serve/src/ingest_api/mod.rs b/quickwit/quickwit-serve/src/ingest_api/mod.rs index 940313fbf6a..3207d7dcc6d 100644 --- a/quickwit/quickwit-serve/src/ingest_api/mod.rs +++ b/quickwit/quickwit-serve/src/ingest_api/mod.rs @@ -16,7 +16,7 @@ mod response; mod rest_handler; pub use response::{RestIngestResponse, RestParseFailure}; +pub(crate) use rest_handler::ingest_api_handlers; #[cfg(test)] pub(crate) use rest_handler::tests::setup_ingest_v1_service; pub use rest_handler::{IngestApi, IngestApiSchemas}; -pub(crate) use rest_handler::{ingest_api_handlers, lines}; diff --git a/quickwit/quickwit-serve/src/ingest_api/rest_handler.rs b/quickwit/quickwit-serve/src/ingest_api/rest_handler.rs index 15db1bf9796..ea91987547f 100644 --- a/quickwit/quickwit-serve/src/ingest_api/rest_handler.rs +++ b/quickwit/quickwit-serve/src/ingest_api/rest_handler.rs @@ -12,17 +12,17 @@ // See the License for the specific language governing permissions and // limitations under the License. -use bytes::{Buf, Bytes}; +use bytes::Buf; use quickwit_config::{INGEST_V2_SOURCE_ID, IngestApiConfig, validate_identifier}; use quickwit_ingest::{ CommitType, DocBatchBuilder, FetchResponse, IngestRequest, IngestService, IngestServiceClient, - IngestServiceError, TailRequest, + IngestServiceError, TailRequest, doc_batch_v2_from_ndjson, split_ndjson_lines, }; +use quickwit_proto::ingest::CommitTypeV2; use quickwit_proto::ingest::router::{ IngestRequestV2, IngestRouterService, IngestRouterServiceClient, IngestSubrequest, }; -use quickwit_proto::ingest::{CommitTypeV2, DocBatchV2}; -use quickwit_proto::types::{DocUidGenerator, IndexId}; +use quickwit_proto::types::IndexId; use serde::Deserialize; use warp::{Filter, Rejection}; @@ -176,7 +176,7 @@ async fn ingest_v1( // The size of the body should be an upper bound of the size of the batch. The removal of the // end of line character for each doc compensates the addition of the `DocCommand` header. let mut doc_batch_builder = DocBatchBuilder::with_capacity(index_id, body.content.remaining()); - for line in lines(&body.content) { + for line in split_ndjson_lines(&body.content) { doc_batch_builder.ingest_doc(line); } let ingest_req = IngestRequest { @@ -193,7 +193,7 @@ async fn ingest_v2( ingest_options: IngestOptions, ingest_router: IngestRouterServiceClient, ) -> Result { - let doc_batch_opt = build_doc_batch_v2_from_ndjson_body(body.content); + let doc_batch_opt = doc_batch_v2_from_ndjson(body.content); let Some(doc_batch) = doc_batch_opt else { let response = RestIngestResponse::default(); @@ -267,65 +267,10 @@ async fn tail_endpoint( Ok(fetch_response) } -pub(crate) fn lines(body: &Bytes) -> impl Iterator { - body.split(|byte| byte == &b'\n') - .filter(|line| !is_empty_or_blank_line(line)) -} - -fn build_doc_batch_v2_from_ndjson_body(ndjson_body: Bytes) -> Option { - let mut doc_uids = Vec::new(); - let mut doc_lengths = Vec::new(); - let mut doc_uid_generator = DocUidGenerator::default(); - let mut segment_start = 0usize; - let mut line_start = 0usize; - - // memchr searches for newlines with SIMD: ~10x faster than a byte-by-byte scan. - for position in memchr::memchr_iter(b'\n', &ndjson_body) { - let line = &ndjson_body[line_start..position]; - if !is_empty_or_blank_line(line) { - doc_uids.push(doc_uid_generator.next_doc_uid()); - doc_lengths.push((position + 1 - segment_start) as u32); - segment_start = position + 1; - } - line_start = position + 1; - } - - let line = &ndjson_body[line_start..]; - if !is_empty_or_blank_line(line) { - doc_uids.push(doc_uid_generator.next_doc_uid()); - doc_lengths.push((ndjson_body.len() - segment_start) as u32); - segment_start = ndjson_body.len(); - } - - if doc_uids.is_empty() { - return None; - } - if segment_start < ndjson_body.len() { - let trailing_whitespace_len = ndjson_body.len() - segment_start; - let last_doc_len = doc_lengths - .last_mut() - .expect("doc lengths should not be empty"); - *last_doc_len += trailing_whitespace_len as u32; - } - - Some(DocBatchV2 { - doc_uids, - doc_buffer: ndjson_body, - doc_lengths, - }) -} - -#[inline] -fn is_empty_or_blank_line(line: &[u8]) -> bool { - line.is_empty() || line.iter().all(|ch| ch.is_ascii_whitespace()) -} - #[cfg(test)] pub(crate) mod tests { - use std::str; use std::time::Duration; - use bytes::Bytes; use quickwit_actors::{Mailbox, Universe}; use quickwit_config::IngestApiConfig; use quickwit_ingest::{ @@ -334,43 +279,7 @@ pub(crate) mod tests { }; use quickwit_proto::ingest::router::IngestRouterServiceClient; - use super::{RestIngestResponse, build_doc_batch_v2_from_ndjson_body, ingest_api_handlers}; - use crate::ingest_api::lines; - - #[test] - fn test_process_lines() { - let test_cases = [ - // an empty line is inserted before the metadata action and the doc - (&b"\n{ \"create\" : { \"_index\" : \"my-index-1\", \"_id\" : \"1\"} }\n{\"id\": 1, \"message\": \"push\"}"[..], 2), - // a blank line is inserted before the metadata action and the doc - (&b" \n{ \"create\" : { \"_index\" : \"my-index-1\", \"_id\" : \"1\"} }\n{\"id\": 1, \"message\": \"push\"}"[..], 2), - // an empty line is inserted after the metadata action and before the doc - (&b"{ \"create\" : { \"_index\" : \"my-index-1\", \"_id\" : \"1\"} }\n\n{\"id\": 1, \"message\": \"push\"}"[..], 2), - // a blank line is inserted after the metadata action and before the doc - (&b"{ \"create\" : { \"_index\" : \"my-index-1\", \"_id\" : \"1\"} }\n \n{\"id\": 1, \"message\": \"push\"}"[..], 2), - ]; - - for &(input, expected_count) in &test_cases { - assert_eq!(lines(&Bytes::from(input)).count(), expected_count); - } - } - - #[test] - fn test_build_doc_batch_v2_from_ndjson_body_zero_copy() { - let body = Bytes::from_static(b"\n {\"id\":1}\n\n{\"id\":2}\n \n"); - let doc_batch = build_doc_batch_v2_from_ndjson_body(body.clone()).unwrap(); - assert_eq!(doc_batch.num_docs(), 2); - assert_eq!(doc_batch.doc_buffer, body); - - let docs: Vec = doc_batch.docs().map(|(_doc_uid, doc)| doc).collect(); - assert_eq!(str::from_utf8(&docs[0]).unwrap(), "\n {\"id\":1}\n"); - assert_eq!(str::from_utf8(&docs[1]).unwrap(), "\n{\"id\":2}\n \n"); - } - - #[test] - fn test_build_doc_batch_v2_from_blank_body() { - assert!(build_doc_batch_v2_from_ndjson_body(Bytes::from_static(b"\n \n\t")).is_none()); - } + use super::{RestIngestResponse, ingest_api_handlers}; pub(crate) async fn setup_ingest_v1_service( queues: &[&str], From 46ec2b6fb0ea4ba3fa412f5b43757c97fc7df19c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fran=C3=A7ois=20Massot?= Date: Sat, 10 Oct 2026 22:03:00 +0200 Subject: [PATCH 6/7] Remove redundant comments --- quickwit/quickwit-ingest/src/ingest_v2/mod.rs | 1 - quickwit/quickwit-ingest/src/ndjson.rs | 18 +++--------------- 2 files changed, 3 insertions(+), 16 deletions(-) diff --git a/quickwit/quickwit-ingest/src/ingest_v2/mod.rs b/quickwit/quickwit-ingest/src/ingest_v2/mod.rs index e77a3dc358d..9b567b82fd7 100644 --- a/quickwit/quickwit-ingest/src/ingest_v2/mod.rs +++ b/quickwit/quickwit-ingest/src/ingest_v2/mod.rs @@ -153,7 +153,6 @@ pub struct DocBatchV2Builder { } impl DocBatchV2Builder { - /// Creates a builder with enough room for `doc_buffer_capacity` bytes of document payloads. pub fn with_capacity(doc_buffer_capacity: usize) -> Self { Self { doc_uids: Vec::new(), diff --git a/quickwit/quickwit-ingest/src/ndjson.rs b/quickwit/quickwit-ingest/src/ndjson.rs index 53b2625425a..402e736a064 100644 --- a/quickwit/quickwit-ingest/src/ndjson.rs +++ b/quickwit/quickwit-ingest/src/ndjson.rs @@ -12,16 +12,11 @@ // See the License for the specific language governing permissions and // limitations under the License. -//! Helpers to split NDJSON request bodies into documents. -//! -//! Newlines are found with `memchr` (SIMD), which is ~10x faster than a byte-by-byte scan on -//! typical log payloads. Empty and blank (whitespace-only) lines are skipped. - use bytes::Bytes; use quickwit_proto::ingest::DocBatchV2; use quickwit_proto::types::DocUidGenerator; -/// Returns an iterator over the non-blank lines of an NDJSON body, without their trailing `\n`. +/// Iterates over the non-blank lines of an NDJSON body. pub fn split_ndjson_lines(ndjson_body: &[u8]) -> impl Iterator { let mut line_start = 0; memchr::memchr_iter(b'\n', ndjson_body) @@ -34,15 +29,8 @@ pub fn split_ndjson_lines(ndjson_body: &[u8]) -> impl Iterator { .filter(|line| !is_empty_or_blank_line(line)) } -/// Builds a [`DocBatchV2`] from an NDJSON body without copying the documents: the body becomes -/// the batch's doc buffer, and each non-blank line becomes a document with a fresh doc UID. -/// -/// Since the docs must cover the whole buffer, the newlines and the blank lines are kept with the -/// neighboring docs: a doc spans from the end of the previous doc to the end of its own line, -/// including the `\n`, and the last doc also covers any trailing blank lines. JSON parsers ignore -/// this leading and trailing whitespace. -/// -/// Returns `None` if the body has no non-blank line. +/// Builds a [`DocBatchV2`] whose doc buffer is the NDJSON body itself, without copying. The docs +/// must cover the whole buffer, so newlines and blank lines stay attached to neighboring docs. pub fn doc_batch_v2_from_ndjson(ndjson_body: Bytes) -> Option { let mut doc_uids = Vec::new(); let mut doc_lengths = Vec::new(); From 32db721ba3f25abb8fbcdcd65767a9ace25a056e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fran=C3=A7ois=20Massot?= Date: Sat, 10 Oct 2026 22:32:21 +0200 Subject: [PATCH 7/7] Account for retained NDJSON newlines in doc-length REST test --- .../rest-api-tests/scenarii/aggregations/0002-doc-len.yaml | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/quickwit/rest-api-tests/scenarii/aggregations/0002-doc-len.yaml b/quickwit/rest-api-tests/scenarii/aggregations/0002-doc-len.yaml index 08b437015b9..3a89f925e75 100644 --- a/quickwit/rest-api-tests/scenarii/aggregations/0002-doc-len.yaml +++ b/quickwit/rest-api-tests/scenarii/aggregations/0002-doc-len.yaml @@ -12,7 +12,8 @@ json: expected: aggregations: doc_len: - value: 952.0 + # Native ingest retains the NDJSON separators: 952 JSON bytes + 10 newlines. + value: 962.0 --- # Test doc len isn't shown when querying documents method: [GET]