From 70dd8eb73f105ac9e6759b9412c0b60397ff4221 Mon Sep 17 00:00:00 2001
From: Pujitha Paladugu <10557236+pujitha24@users.noreply.github.com>
Date: Fri, 16 Oct 2026 22:29:06 -0700
Subject: [PATCH] fix(rest-client): only send the commit mode with the last
 ingest batch

Motivation: `quickwit index ingest --force` documents that the commit is
forced after the last document is sent. The REST client added the
`commit=force` (or `wait_for`) query parameter to every batch request
instead, so each batch (e.g. ~8 MB with --batch-size-limit 8MB) forced a
commit and published its own small split, causing extra uploads,
metastore operations and merge work.

Approach: in QuickwitClient::ingest, set the `commit` query parameter
only when the batch is the last one. Earlier batches use the default
(auto) commit. The commit timeout was already applied only to the last
batch, so both now use the same condition.

Validation: added test_ingest_endpoint_force_commit_only_on_last_batch,
which ingests three single-document batches with CommitType::Force and
asserts the commit parameter is [None, None, "force"]. With the fix
disabled it fails (all three requests carried commit=force); with the fix
it passes. `cargo test -p quickwit-rest-client` passes (12 tests) and
`cargo clippy -p quickwit-rest-client --tests` is clean. I did not run
an end-to-end ingest against a live node; the change is confined to
request construction in the client, which the test covers.

Report: https://github.com/quickwit-oss/quickwit/issues/6615
Signed-off-by: Pujitha Paladugu <10557236+pujitha24@users.noreply.github.com>
Assisted-by: claude-sonnet-5-5 (via Claude Code)
---
 .../quickwit-rest-client/src/rest_client.rs   | 59 ++++++++++++++++---
 1 file changed, 51 insertions(+), 8 deletions(-)

diff --git a/quickwit/quickwit-rest-client/src/rest_client.rs b/quickwit/quickwit-rest-client/src/rest_client.rs
index ea5eda5a391..da5be00cb05 100644
--- a/quickwit/quickwit-rest-client/src/rest_client.rs
+++ b/quickwit/quickwit-rest-client/src/rest_client.rs
@@ -343,18 +343,23 @@ impl QuickwitClient {
         let mut cumulated_resp = RestIngestResponse::default();
         while let Some(batch) = batch_reader.next_batch().await? {
             loop {
-                let timeout = if !batch_reader.has_next() && last_block_commit != CommitType::Auto {
+                // Only the last batch carries the requested commit mode; the other batches
+                // use the default (auto) commit.
+                let is_last_batch = !batch_reader.has_next();
+                let timeout = if is_last_batch && last_block_commit != CommitType::Auto {
                     self.commit_timeout
                 } else {
                     self.ingest_timeout
                 };
-                match last_block_commit {
-                    CommitType::Auto => {}
-                    CommitType::WaitFor => {
-                        query_params.insert("commit", "wait_for");
-                    }
-                    CommitType::Force => {
-                        query_params.insert("commit", "force");
+                if is_last_batch {
+                    match last_block_commit {
+                        CommitType::Auto => {}
+                        CommitType::WaitFor => {
+                            query_params.insert("commit", "wait_for");
+                        }
+                        CommitType::Force => {
+                            query_params.insert("commit", "force");
+                        }
                     }
                 }
                 let response = self
@@ -961,6 +966,44 @@ mod test {
         assert_eq!(actual_response, mock_response);
     }
 
+    #[tokio::test]
+    async fn test_ingest_endpoint_force_commit_only_on_last_batch() {
+        let mock_server = MockServer::start().await;
+        let server_url = Url::parse(&mock_server.uri()).unwrap();
+        let qw_client = QuickwitClientBuilder::new(server_url).build();
+        let mock_response = RestIngestResponse {
+            num_docs_for_processing: 1,
+            num_ingested_docs: Some(1),
+            num_rejected_docs: Some(0),
+            parse_failures: Some(Vec::new()),
+        };
+        Mock::given(method("POST"))
+            .and(path("/api/v1/my-index/ingest"))
+            .respond_with(ResponseTemplate::new(StatusCode::OK).set_body_json(&mock_response))
+            .mount(&mock_server)
+            .await;
+        // Each document is 10 bytes long, so a 12-byte limit yields one batch per document.
+        let ingest_source =
+            IngestSource::Str("{\"id\": 1}\n{\"id\": 2}\n{\"id\": 3}\n".to_string());
+        qw_client
+            .ingest("my-index", ingest_source, Some(12), None, CommitType::Force)
+            .await
+            .unwrap();
+        let requests = mock_server.received_requests().await.unwrap();
+        assert_eq!(requests.len(), 3);
+        let commit_params: Vec<Option<String>> = requests
+            .iter()
+            .map(|request| {
+                request
+                    .url
+                    .query_pairs()
+                    .find(|(key, _)| key == "commit")
+                    .map(|(_, value)| value.to_string())
+            })
+            .collect();
+        assert_eq!(commit_params, vec![None, None, Some("force".to_string())]);
+    }
+
     #[tokio::test]
     async fn test_ingest_endpoint_with_wait_for_commit() {
         let mock_server = MockServer::start().await;