searchengines-web/MeiliSearch
searchengines-web/MeiliSearch
1
0
Fork 0
mirror of https://github.com/meilisearch/MeiliSearch synced 2025-07-03 11:57:07 +02:00
Code Issues Releases Wiki Activity

Gzip-compress the content

Browse source
This commit is contained in:
Kerollmops 2025-06-25 12:29:14 +02:00
parent c6216517c7
commit a743da3061
No known key found for this signature in database
GPG key ID: F250A4C4E3AE5F5F
1 changed files with 20 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

23
crates/index-scheduler/src/scheduler/process_export.rs
View file

@ -1,9 +1,11 @@
use std::collections::BTreeMap; use std::collections::BTreeMap;
use std::io; use std::io::{self, Write as _};
use std::sync::atomic; use std::sync::atomic;
use std::time::Duration; use std::time::Duration;
use backoff::ExponentialBackoff; use backoff::ExponentialBackoff;
use flate2::write::GzEncoder;
use flate2::Compression;
use meilisearch_types::index_uid_pattern::IndexUidPattern; use meilisearch_types::index_uid_pattern::IndexUidPattern;
use meilisearch_types::milli::constants::RESERVED_VECTORS_FIELD_NAME; use meilisearch_types::milli::constants::RESERVED_VECTORS_FIELD_NAME;
use meilisearch_types::milli::progress::{Progress, VariableNameStep}; use meilisearch_types::milli::progress::{Progress, VariableNameStep};
@ -131,6 +133,7 @@ impl IndexScheduler {
let mut buffer = Vec::new(); let mut buffer = Vec::new();
let mut tmp_buffer = Vec::new(); let mut tmp_buffer = Vec::new();
let mut compressed_buffer = Vec::new();
for (i, docid) in universe.iter().enumerate() { for (i, docid) in universe.iter().enumerate() {
if i % ctx.num_threads() != ctx.index() { if i % ctx.num_threads() != ctx.index() {
continue; continue;
@ -205,17 +208,31 @@ impl IndexScheduler {
.map_err(milli::InternalError::from) .map_err(milli::InternalError::from)
.map_err(|e| Error::from_milli(e.into(), Some(uid.to_string())))?; .map_err(|e| Error::from_milli(e.into(), Some(uid.to_string())))?;
if buffer.len() + tmp_buffer.len() > limit { // Make sure we put at least one document in the buffer even
// though we might go above the buffer limit before sending
if !buffer.is_empty() && buffer.len() + tmp_buffer.len() > limit {
// We compress the documents before sending them
let mut encoder =
GzEncoder::new(&mut compressed_buffer, Compression::default());
encoder
.write_all(&buffer)
.map_err(|e| Error::from_milli(e.into(), Some(uid.clone())))?;
encoder
.finish()
.map_err(|e| Error::from_milli(e.into(), Some(uid.clone())))?;
retry(&must_stop_processing, || { retry(&must_stop_processing, || {
let mut request = agent.post(&documents_url); let mut request = agent.post(&documents_url);
request = request.set("Content-Type", "application/x-ndjson"); request = request.set("Content-Type", "application/x-ndjson");
request = request.set("Content-Encoding", "gzip");
if let Some(api_key) = api_key { if let Some(api_key) = api_key {
request = request request = request
.set("Authorization", &(format!("Bearer {api_key}"))); .set("Authorization", &(format!("Bearer {api_key}")));
} }
request.send_bytes(&buffer).map_err(into_backoff_error) request.send_bytes(&compressed_buffer).map_err(into_backoff_error)
})?; })?;
buffer.clear(); buffer.clear();
compressed_buffer.clear();
} }
buffer.extend_from_slice(&tmp_buffer); buffer.extend_from_slice(&tmp_buffer);