test mrecordlog

quickwit/Cargo.toml

@@ -153,7 +153,7 @@ matches = "0.1.9"
 md5 = "0.7"
 mime_guess = "2.0.4"
 mockall = "0.11"
-mrecordlog = { git = "https://github.com/quickwit-oss/mrecordlog", rev = "187486f" }
+mrecordlog = { git = "https://github.com/quickwit-oss/mrecordlog", rev = "bdd759f9b0" }
 new_string_template = "1.4.0"
 nom = "7.1.3"
 num_cpus = "1"

quickwit/quickwit-ingest/src/ingest_v2/fetch.rs

@@ -17,13 +17,12 @@
 // 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::borrow::Borrow;
 use std::collections::hash_map::Entry;
 use std::collections::HashMap;
 use std::fmt;
 use std::sync::Arc;
 
-use bytes::{BufMut, BytesMut};
+use bytes::{Buf, BufMut, BytesMut};
 use futures::StreamExt;
 use mrecordlog::Record;
 use quickwit_common::retry::RetryParams;
@@ -143,12 +142,12 @@ impl FetchStreamTask {
                 break;
             };
             for Record { payload, .. } in mrecords {
-                if mrecord_buffer.len() + payload.len() > mrecord_buffer.capacity() {
+                if mrecord_buffer.len() + payload.remaining() > mrecord_buffer.capacity() {
                     has_drained_queue = false;
                     break;
                 }
-                mrecord_buffer.put(payload.borrow());
-                mrecord_lengths.push(payload.len() as u32);
+                mrecord_lengths.push(payload.remaining() as u32);
+                mrecord_buffer.put(payload);
             }
             // Drop the lock while we send the message.
             drop(mrecordlog_guard);

quickwit/quickwit-ingest/src/queue.rs

@@ -183,7 +183,7 @@ impl Queues {
             if first_key_opt.is_none() {
                 first_key_opt = Some(position);
             }
-            num_bytes += doc_batch.command_from_buf(payload.as_ref());
+            num_bytes += doc_batch.command_from_buf(payload);
             if num_bytes > size_limit {
                 break;
             }