mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-26 08:18:18 +00:00
refactor(zip): simplify archive extraction path (#3290)
* refactor(zip): simplify archive extraction path * build(deps): align datafusion and http updates * fix(test): restore pre-commit compatibility
This commit is contained in:
Generated
+162
-239
File diff suppressed because it is too large
Load Diff
+4
-4
@@ -136,7 +136,7 @@ lapin = { version = "4.10.0", default-features = false, features = ["tokio", "ru
|
||||
hyper = { version = "1.10.1", features = ["http2", "http1", "server"] }
|
||||
hyper-rustls = { version = "0.27.9", default-features = false, features = ["native-tokio", "http1", "tls12", "logging", "http2", "aws-lc-rs", "webpki-roots"] }
|
||||
hyper-util = { version = "0.1.20", features = ["tokio", "server-auto", "server-graceful", "tracing"] }
|
||||
http = "1.4.1"
|
||||
http = "1.4.2"
|
||||
http-body = "1.0.1"
|
||||
http-body-util = "0.1.3"
|
||||
reqwest = { version = "0.13.4", default-features = false, features = ["rustls", "charset", "http2", "system-proxy", "stream", "json", "blocking", "query", "form"] }
|
||||
@@ -219,7 +219,7 @@ crossbeam-queue = "0.3.12"
|
||||
crossbeam-channel = "0.5.15"
|
||||
crossbeam-deque = "0.8.6"
|
||||
crossbeam-utils = "0.8.21"
|
||||
datafusion = "53.1.0"
|
||||
datafusion = "54.0.0"
|
||||
derive_builder = "0.20.2"
|
||||
enumset = "1.1.13"
|
||||
faster-hex = "0.10.0"
|
||||
@@ -310,9 +310,9 @@ pyroscope = { version = "=2.0.5", features = ["backend-pprof-rs", "backend-jemal
|
||||
# FTP and SFTP
|
||||
libunftp = { version = "0.23.0", features = ["experimental"] }
|
||||
unftp-core = "0.1.0"
|
||||
suppaftp = { version = "8.0.3", features = ["tokio", "tokio-rustls-aws-lc-rs"] }
|
||||
suppaftp = { version = "8.0.4", features = ["tokio", "tokio-rustls-aws-lc-rs"] }
|
||||
rcgen = "0.14.8"
|
||||
russh = { version = "0.61.2",features = ["serde"] }
|
||||
russh = { version = "0.61.2", features = ["serde"] }
|
||||
russh-sftp = "2.3.0"
|
||||
|
||||
# WebDAV
|
||||
|
||||
@@ -12,7 +12,6 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use std::any::Any;
|
||||
use std::borrow::Cow;
|
||||
use std::fmt::Display;
|
||||
use std::sync::Arc;
|
||||
@@ -78,10 +77,6 @@ impl TableSourceAdapter {
|
||||
|
||||
#[async_trait]
|
||||
impl TableSource for TableSourceAdapter {
|
||||
fn as_any(&self) -> &dyn Any {
|
||||
self
|
||||
}
|
||||
|
||||
fn schema(&self) -> SchemaRef {
|
||||
self.table_handle.schema()
|
||||
}
|
||||
|
||||
@@ -12,7 +12,7 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use std::{any::Any, fmt, sync::Arc};
|
||||
use std::{fmt, sync::Arc};
|
||||
|
||||
use async_trait::async_trait;
|
||||
use datafusion::{
|
||||
@@ -88,8 +88,7 @@ impl ParquetSelectTable {
|
||||
fn partitioned_file(&self) -> PartitionedFile {
|
||||
let file = PartitionedFile::new(self.object_path.clone(), self.object_size);
|
||||
if let Some(access_plan) = self.access_plan.as_ref() {
|
||||
let extensions: Arc<dyn Any + Send + Sync> = access_plan.clone();
|
||||
file.with_extensions(extensions)
|
||||
file.with_extension(access_plan.as_ref().clone())
|
||||
} else {
|
||||
file
|
||||
}
|
||||
@@ -98,10 +97,6 @@ impl ParquetSelectTable {
|
||||
|
||||
#[async_trait]
|
||||
impl TableProvider for ParquetSelectTable {
|
||||
fn as_any(&self) -> &dyn Any {
|
||||
self
|
||||
}
|
||||
|
||||
fn schema(&self) -> SchemaRef {
|
||||
Arc::clone(&self.schema)
|
||||
}
|
||||
|
||||
@@ -19,7 +19,7 @@ use datafusion::arrow::datatypes::DataType;
|
||||
use datafusion::common::Result as DFResult;
|
||||
use datafusion::datasource::TableProvider;
|
||||
use datafusion::logical_expr::var_provider::is_system_variables;
|
||||
use datafusion::logical_expr::{AggregateUDF, ScalarUDF, TableSource, WindowUDF};
|
||||
use datafusion::logical_expr::{AggregateUDF, HigherOrderUDF, ScalarUDF, TableSource, WindowUDF};
|
||||
use datafusion::variable::VarType;
|
||||
use datafusion::{
|
||||
config::ConfigOptions,
|
||||
@@ -95,6 +95,10 @@ impl ContextProvider for MetadataProvider {
|
||||
self.func_manager.udaf(name).ok()
|
||||
}
|
||||
|
||||
fn get_higher_order_meta(&self, _name: &str) -> Option<Arc<HigherOrderUDF>> {
|
||||
None
|
||||
}
|
||||
|
||||
fn get_variable_type(&self, variable_names: &[String]) -> Option<DataType> {
|
||||
if variable_names.is_empty() {
|
||||
return None;
|
||||
@@ -130,6 +134,10 @@ impl ContextProvider for MetadataProvider {
|
||||
self.func_manager.udfs()
|
||||
}
|
||||
|
||||
fn higher_order_function_names(&self) -> Vec<String> {
|
||||
Vec::new()
|
||||
}
|
||||
|
||||
fn udaf_names(&self) -> Vec<String> {
|
||||
self.func_manager.udafs()
|
||||
}
|
||||
|
||||
+12
-3
@@ -25,6 +25,12 @@ keywords = ["zip", "compression", "rustfs", "Minio"]
|
||||
categories = ["web-programming", "development-tools", "compression"]
|
||||
documentation = "https://docs.rs/rustfs-zip/latest/rustfs_zip/"
|
||||
|
||||
[lib]
|
||||
doctest = false
|
||||
|
||||
[[bench]]
|
||||
name = "zip_benchmark"
|
||||
harness = false
|
||||
|
||||
[dependencies]
|
||||
async-compression = { workspace = true, features = [
|
||||
@@ -38,10 +44,13 @@ async-compression = { workspace = true, features = [
|
||||
tokio = { workspace = true, features = ["io-uring","fs","io-util","macros"] }
|
||||
tokio-stream = { workspace = true }
|
||||
astral-tokio-tar = { workspace = true }
|
||||
thiserror = { workspace = true }
|
||||
zip = { workspace = true }
|
||||
|
||||
[dev-dependencies]
|
||||
criterion = { workspace = true, features = ["html_reports"] }
|
||||
tempfile = { workspace = true }
|
||||
|
||||
|
||||
[lints]
|
||||
workspace = true
|
||||
|
||||
[lib]
|
||||
doctest = false
|
||||
|
||||
+47
-9
@@ -1,6 +1,6 @@
|
||||
[](https://rustfs.com)
|
||||
|
||||
# RustFS Zip - Compression & Archiving
|
||||
# RustFS Zip - Archive And Compression Primitives
|
||||
|
||||
<p align="center">
|
||||
<strong>High-performance compression and archiving for RustFS object storage</strong>
|
||||
@@ -17,16 +17,54 @@
|
||||
|
||||
## 📖 Overview
|
||||
|
||||
**RustFS Zip** provides high-performance compression and archiving capabilities for the [RustFS](https://rustfs.com) distributed object storage system. For the complete RustFS experience, please visit the [main RustFS repository](https://github.com/rustfs/rustfs).
|
||||
**RustFS Zip** provides archive and compression primitives for the [RustFS](https://rustfs.com) distributed object storage system. Today it is primarily used by RustFS archive extract flows to:
|
||||
|
||||
## ✨ Features
|
||||
- identify archive/compression formats by extension
|
||||
- stream tar and tar+compression inputs through async decoders
|
||||
- provide small ZIP read/write helpers for local archive workflows
|
||||
|
||||
- Multiple compression algorithms (Zstd, LZ4, Gzip, Brotli)
|
||||
- Streaming compression for memory efficiency
|
||||
- Parallel processing for large files
|
||||
- Archive format support (ZIP, TAR, custom formats)
|
||||
- Adaptive compression with content-type detection
|
||||
- Compression analytics and performance metrics
|
||||
## Current Features
|
||||
|
||||
- A clearer type model with:
|
||||
- `CompressionCodec` for stream codecs
|
||||
- `ArchiveKind` for container families
|
||||
- `ArchiveFormat` for concrete archive/container combinations
|
||||
- Async stream codecs for `gzip`, `bzip2`, `zlib`, `xz`, and `zstd`
|
||||
- Tar archive iteration over async readers through `read_archive_entries()` / `extract_tar_entries()`
|
||||
- Archive guardrails through `ArchiveLimits` for entry count, entry size, total unpacked size, and path length
|
||||
- In-memory compression helpers for payload round-trip workflows
|
||||
- Blocking ZIP create/extract helpers for local archive files
|
||||
- ZIP helper metadata via `ZipEntry`, including:
|
||||
- `compression_method`
|
||||
- `archive_kind`
|
||||
- `format`
|
||||
- `unix_mode`
|
||||
- ZIP helper options via `ZipWriteOptions`, including:
|
||||
- `compression_level`
|
||||
- `create_directory_entries`
|
||||
|
||||
## Compatibility
|
||||
|
||||
- `CompressionFormat` is retained as a compatibility layer for existing callers
|
||||
- New code should prefer `ArchiveFormat`, `ArchiveKind`, and `CompressionCodec` when expressing archive semantics
|
||||
|
||||
## ZIP Helper Scope
|
||||
|
||||
The file-based ZIP helper APIs are best suited for:
|
||||
|
||||
- local archive import/export flows
|
||||
- admin-side packaging helpers
|
||||
- test fixtures and tooling
|
||||
|
||||
They are not intended to be a remote streaming ZIP access engine.
|
||||
|
||||
## Current Boundaries
|
||||
|
||||
- ZIP is supported via file-based helper APIs, not the tar-family async stream APIs
|
||||
- Tar-family stream APIs are intended for `tar`, `tar.gz`, `tar.bz2`, `tar.xz`, `tar.zst`, and similar compressed tar flows
|
||||
- Default archive guardrails are intentionally conservative and do not replace higher-level RustFS object-path validation
|
||||
- This crate does not currently implement a general-purpose parallel archive engine
|
||||
- Archive extraction safety policy remains the responsibility of the RustFS caller for object-store flows
|
||||
|
||||
## 📚 Documentation
|
||||
|
||||
|
||||
@@ -0,0 +1,416 @@
|
||||
// Copyright 2024 RustFS Team
|
||||
//
|
||||
// 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.
|
||||
|
||||
use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
|
||||
use rustfs_zip::{
|
||||
ArchiveLimits, CompressionFormat, CompressionLevel, ZipWriteOptions, create_zip_with_options, extract_tar_entries,
|
||||
extract_zip_to_path_with_limits, extract_zip_with_limits,
|
||||
};
|
||||
use std::hint::black_box;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use tempfile::tempdir;
|
||||
use tokio::runtime::Builder;
|
||||
use tokio_tar::{Builder as TarBuilder, Header};
|
||||
use zip::ZipArchive;
|
||||
|
||||
fn build_runtime() -> tokio::runtime::Runtime {
|
||||
Builder::new_current_thread()
|
||||
.enable_all()
|
||||
.build()
|
||||
.expect("build tokio runtime for rustfs-zip benchmarks")
|
||||
}
|
||||
|
||||
async fn build_tar_payload(entry_count: usize, payload_size: usize) -> Vec<u8> {
|
||||
let sink = tokio::io::duplex(64 * 1024);
|
||||
let (writer, mut reader) = sink;
|
||||
let write_task = tokio::spawn(async move {
|
||||
let mut builder = TarBuilder::new(writer);
|
||||
let payload = vec![b'a'; payload_size];
|
||||
for index in 0..entry_count {
|
||||
let mut header = Header::new_gnu();
|
||||
header.set_size(payload.len() as u64);
|
||||
header.set_mode(0o644);
|
||||
header.set_cksum();
|
||||
builder
|
||||
.append_data(&mut header, format!("entry-{index}.txt"), &payload[..])
|
||||
.await
|
||||
.expect("append tar benchmark entry");
|
||||
}
|
||||
builder.finish().await.expect("finish tar benchmark archive");
|
||||
});
|
||||
|
||||
let mut output = Vec::new();
|
||||
tokio::io::copy(&mut reader, &mut output)
|
||||
.await
|
||||
.expect("read tar benchmark archive");
|
||||
write_task.await.expect("join tar writer task");
|
||||
output
|
||||
}
|
||||
|
||||
async fn build_compressed_tar_payload(format: CompressionFormat, entry_count: usize, payload_size: usize) -> Vec<u8> {
|
||||
let tar_payload = build_tar_payload(entry_count, payload_size).await;
|
||||
rustfs_zip::Compressor::new(format)
|
||||
.compress(&tar_payload)
|
||||
.await
|
||||
.expect("compress tar benchmark payload")
|
||||
}
|
||||
|
||||
fn bench_tar_family_extract(c: &mut Criterion) {
|
||||
let runtime = build_runtime();
|
||||
let mut group = c.benchmark_group("zip_tar_family_extract");
|
||||
|
||||
for (name, format, entry_count, payload_size) in [
|
||||
("tar_gzip_small_many", CompressionFormat::Gzip, 64usize, 256usize),
|
||||
("tar_zstd_medium", CompressionFormat::Zstd, 16usize, 16 * 1024usize),
|
||||
] {
|
||||
let payload = runtime.block_on(build_compressed_tar_payload(format, entry_count, payload_size));
|
||||
group.throughput(Throughput::Bytes(payload.len() as u64));
|
||||
group.bench_with_input(BenchmarkId::new(name, payload.len()), &payload, |b, payload| {
|
||||
b.iter(|| {
|
||||
runtime.block_on(async {
|
||||
let seen = Arc::new(AtomicUsize::new(0));
|
||||
let seen_ref = Arc::clone(&seen);
|
||||
extract_tar_entries(std::io::Cursor::new(payload.clone()), format, move |_entry| {
|
||||
let seen_ref = Arc::clone(&seen_ref);
|
||||
async move {
|
||||
seen_ref.fetch_add(1, Ordering::Relaxed);
|
||||
Ok(())
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("extract tar benchmark payload");
|
||||
black_box(seen.load(Ordering::Relaxed));
|
||||
});
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
group.finish();
|
||||
}
|
||||
|
||||
fn bench_zip_helper_round_trip(c: &mut Criterion) {
|
||||
let runtime = build_runtime();
|
||||
let mut group = c.benchmark_group("zip_helper_round_trip");
|
||||
|
||||
let zip_matrix = [
|
||||
("stored_flat_32x128", CompressionLevel::Fastest, 32usize, 128usize, "flat"),
|
||||
("stored_nested_32x256", CompressionLevel::Fastest, 32usize, 256usize, "nested"),
|
||||
("stored_flat_256x128", CompressionLevel::Fastest, 256usize, 128usize, "flat"),
|
||||
("deflated_flat_32x1k", CompressionLevel::Best, 32usize, 1024usize, "flat"),
|
||||
("deflated_nested_256x1k", CompressionLevel::Best, 256usize, 1024usize, "nested"),
|
||||
("deflated_deep_1024x4k", CompressionLevel::Best, 1024usize, 4 * 1024usize, "deep"),
|
||||
];
|
||||
|
||||
for (name, compression_level, file_count, payload_size, layout) in zip_matrix {
|
||||
let files = (0..file_count)
|
||||
.map(|index| {
|
||||
let path = match layout {
|
||||
"flat" => format!("file-{index}.txt"),
|
||||
"nested" => format!("batch-{}/file-{index}.txt", index % 8),
|
||||
"deep" => format!("lvl1/lvl2-{}/lvl3-{}/file-{index}.txt", index % 16, index % 32),
|
||||
_ => format!("file-{index}.txt"),
|
||||
};
|
||||
(path, vec![b'b'; payload_size])
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
let total_bytes = (file_count * payload_size) as u64;
|
||||
group.throughput(Throughput::Bytes(total_bytes));
|
||||
|
||||
group.bench_with_input(BenchmarkId::new(name, total_bytes), &files, |b, files| {
|
||||
b.iter(|| {
|
||||
let temp = tempdir().expect("create benchmark tempdir");
|
||||
let zip_path = temp.path().join("archive.zip");
|
||||
let extract_path = temp.path().join("extract");
|
||||
runtime.block_on(async {
|
||||
create_zip_with_options(
|
||||
&zip_path,
|
||||
files.clone(),
|
||||
ZipWriteOptions {
|
||||
compression_level,
|
||||
create_directory_entries: true,
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("create zip benchmark archive");
|
||||
|
||||
let entries = extract_zip_with_limits(&zip_path, &extract_path, ArchiveLimits::default())
|
||||
.await
|
||||
.expect("extract zip benchmark archive");
|
||||
black_box(entries.len());
|
||||
});
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
group.finish();
|
||||
}
|
||||
|
||||
fn bench_zip_helper_hotspot_breakdown(c: &mut Criterion) {
|
||||
let runtime = build_runtime();
|
||||
let mut group = c.benchmark_group("zip_helper_hotspot_breakdown");
|
||||
let files = (0..32)
|
||||
.map(|index| (format!("batch/file-{index}.txt"), vec![b'c'; 256]))
|
||||
.collect::<Vec<_>>();
|
||||
let total_bytes = (32 * 256) as u64;
|
||||
group.throughput(Throughput::Bytes(total_bytes));
|
||||
|
||||
group.bench_function("fs_setup_cleanup_only", |b| {
|
||||
b.iter(|| {
|
||||
let temp = tempdir().expect("create benchmark tempdir");
|
||||
let zip_path = temp.path().join("archive.zip");
|
||||
let extract_path = temp.path().join("extract");
|
||||
black_box((zip_path, extract_path));
|
||||
});
|
||||
});
|
||||
|
||||
group.bench_function("zip_create_only_stored_small", |b| {
|
||||
b.iter(|| {
|
||||
let temp = tempdir().expect("create benchmark tempdir");
|
||||
let zip_path = temp.path().join("archive.zip");
|
||||
runtime.block_on(async {
|
||||
create_zip_with_options(
|
||||
&zip_path,
|
||||
files.clone(),
|
||||
ZipWriteOptions {
|
||||
compression_level: CompressionLevel::Fastest,
|
||||
create_directory_entries: true,
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("create zip benchmark archive");
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
let payload_for_extract = {
|
||||
let temp = tempdir().expect("create benchmark tempdir");
|
||||
let zip_path = temp.path().join("archive.zip");
|
||||
runtime.block_on(async {
|
||||
create_zip_with_options(
|
||||
&zip_path,
|
||||
files.clone(),
|
||||
ZipWriteOptions {
|
||||
compression_level: CompressionLevel::Fastest,
|
||||
create_directory_entries: true,
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("prepare zip benchmark extract payload");
|
||||
});
|
||||
std::fs::read(&zip_path).expect("read benchmark zip payload")
|
||||
};
|
||||
|
||||
group.bench_function("zip_extract_only_stored_small", |b| {
|
||||
b.iter(|| {
|
||||
let temp = tempdir().expect("create benchmark tempdir");
|
||||
let zip_path = temp.path().join("archive.zip");
|
||||
let extract_path = temp.path().join("extract");
|
||||
std::fs::write(&zip_path, &payload_for_extract).expect("write benchmark zip payload");
|
||||
runtime.block_on(async {
|
||||
let entries = extract_zip_with_limits(&zip_path, &extract_path, ArchiveLimits::default())
|
||||
.await
|
||||
.expect("extract zip benchmark archive");
|
||||
black_box(entries.len());
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
group.bench_function("zip_extract_only_stored_small_summary_only", |b| {
|
||||
b.iter(|| {
|
||||
let temp = tempdir().expect("create benchmark tempdir");
|
||||
let zip_path = temp.path().join("archive.zip");
|
||||
let extract_path = temp.path().join("extract");
|
||||
std::fs::write(&zip_path, &payload_for_extract).expect("write benchmark zip payload");
|
||||
runtime.block_on(async {
|
||||
let summary = extract_zip_to_path_with_limits(&zip_path, &extract_path, ArchiveLimits::default())
|
||||
.await
|
||||
.expect("extract zip benchmark summary path");
|
||||
black_box(summary.entry_count);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
group.bench_function("zip_reader_only_stored_small", |b| {
|
||||
b.iter(|| {
|
||||
let cursor = std::io::Cursor::new(payload_for_extract.clone());
|
||||
let mut archive = ZipArchive::new(cursor).expect("open zip archive for reader-only benchmark");
|
||||
let mut total_bytes = 0usize;
|
||||
for index in 0..archive.len() {
|
||||
let mut zip_file = archive.by_index(index).expect("access zip entry by index");
|
||||
let enclosed_name = zip_file
|
||||
.enclosed_name()
|
||||
.expect("resolve enclosed zip entry name")
|
||||
.to_string_lossy()
|
||||
.replace('\\', "/");
|
||||
let size = zip_file.size();
|
||||
assert!(!enclosed_name.is_empty(), "zip reader-only benchmark expects non-empty names");
|
||||
assert!(
|
||||
size <= ArchiveLimits::default().max_entry_size,
|
||||
"zip reader-only benchmark expects small entries"
|
||||
);
|
||||
if !zip_file.is_dir() {
|
||||
let mut sink = [0_u8; 256];
|
||||
let bytes_read =
|
||||
std::io::Read::read(&mut zip_file, &mut sink).expect("read zip entry payload for reader-only benchmark");
|
||||
total_bytes += bytes_read;
|
||||
}
|
||||
}
|
||||
black_box(total_bytes);
|
||||
});
|
||||
});
|
||||
|
||||
group.bench_function("file_write_only_stored_small", |b| {
|
||||
b.iter(|| {
|
||||
let temp = tempdir().expect("create benchmark tempdir");
|
||||
let extract_path = temp.path().join("extract");
|
||||
std::fs::create_dir_all(&extract_path).expect("create extract dir for file-write-only benchmark");
|
||||
let mut total_bytes = 0usize;
|
||||
for index in 0..32 {
|
||||
let path = extract_path.join(format!("file-{index}.txt"));
|
||||
std::fs::write(&path, [b'c'; 256]).expect("write small file for file-write-only benchmark");
|
||||
total_bytes += 256;
|
||||
}
|
||||
black_box(total_bytes);
|
||||
});
|
||||
});
|
||||
|
||||
group.finish();
|
||||
}
|
||||
|
||||
fn build_object_archive_files(
|
||||
metadata_count: usize,
|
||||
metadata_size: usize,
|
||||
payload_count: usize,
|
||||
payload_size: usize,
|
||||
) -> Vec<(String, Vec<u8>)> {
|
||||
let mut files = Vec::with_capacity(metadata_count * 2 + payload_count);
|
||||
|
||||
for index in 0..metadata_count {
|
||||
let key_prefix = format!(
|
||||
"bucket-a/shard-{}/tenant-{}/dataset-{}/object-{index:04}",
|
||||
index % 8,
|
||||
index % 16,
|
||||
index % 32
|
||||
);
|
||||
files.push((
|
||||
format!("{key_prefix}/meta.json"),
|
||||
format!(
|
||||
"{{\"key\":\"object-{index:04}\",\"etag\":\"{:032x}\",\"size\":{},\"content_type\":\"application/octet-stream\"}}",
|
||||
index,
|
||||
payload_size
|
||||
)
|
||||
.into_bytes(),
|
||||
));
|
||||
files.push((format!("{key_prefix}/tags.txt"), vec![b'm'; metadata_size]));
|
||||
}
|
||||
|
||||
for index in 0..payload_count {
|
||||
let payload_prefix = format!(
|
||||
"bucket-a/shard-{}/tenant-{}/dataset-{}/object-{index:04}",
|
||||
index % 8,
|
||||
index % 16,
|
||||
index % 32
|
||||
);
|
||||
files.push((format!("{payload_prefix}/part-00000.bin"), vec![b'p'; payload_size]));
|
||||
}
|
||||
|
||||
files
|
||||
}
|
||||
|
||||
fn bench_zip_object_archive_extract(c: &mut Criterion) {
|
||||
let runtime = build_runtime();
|
||||
let mut group = c.benchmark_group("zip_object_archive_extract");
|
||||
|
||||
for (name, compression_level, metadata_count, metadata_size, payload_count, payload_size) in [
|
||||
(
|
||||
"stored_metadata_heavy_384m_24p",
|
||||
CompressionLevel::Fastest,
|
||||
384usize,
|
||||
192usize,
|
||||
24usize,
|
||||
32 * 1024usize,
|
||||
),
|
||||
(
|
||||
"deflated_mixed_192m_32p",
|
||||
CompressionLevel::Best,
|
||||
192usize,
|
||||
256usize,
|
||||
32usize,
|
||||
64 * 1024usize,
|
||||
),
|
||||
] {
|
||||
let files = build_object_archive_files(metadata_count, metadata_size, payload_count, payload_size);
|
||||
let total_bytes = files.iter().map(|(_, payload)| payload.len() as u64).sum::<u64>();
|
||||
let payload = {
|
||||
let temp = tempdir().expect("create benchmark tempdir");
|
||||
let zip_path = temp.path().join("object-archive.zip");
|
||||
runtime.block_on(async {
|
||||
create_zip_with_options(
|
||||
&zip_path,
|
||||
files.clone(),
|
||||
ZipWriteOptions {
|
||||
compression_level,
|
||||
create_directory_entries: true,
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("create object archive benchmark payload");
|
||||
});
|
||||
std::fs::read(&zip_path).expect("read object archive benchmark payload")
|
||||
};
|
||||
|
||||
group.throughput(Throughput::Bytes(total_bytes));
|
||||
group.bench_function(BenchmarkId::new("extract_full", name), |b| {
|
||||
b.iter(|| {
|
||||
let temp = tempdir().expect("create benchmark tempdir");
|
||||
let zip_path = temp.path().join("archive.zip");
|
||||
let extract_path = temp.path().join("extract");
|
||||
std::fs::write(&zip_path, &payload).expect("write object archive benchmark payload");
|
||||
runtime.block_on(async {
|
||||
let entries = extract_zip_with_limits(&zip_path, &extract_path, ArchiveLimits::default())
|
||||
.await
|
||||
.expect("extract object archive benchmark payload");
|
||||
black_box(entries.len());
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
group.bench_function(BenchmarkId::new("extract_summary_only", name), |b| {
|
||||
b.iter(|| {
|
||||
let temp = tempdir().expect("create benchmark tempdir");
|
||||
let zip_path = temp.path().join("archive.zip");
|
||||
let extract_path = temp.path().join("extract");
|
||||
std::fs::write(&zip_path, &payload).expect("write object archive benchmark payload");
|
||||
runtime.block_on(async {
|
||||
let summary = extract_zip_to_path_with_limits(&zip_path, &extract_path, ArchiveLimits::default())
|
||||
.await
|
||||
.expect("extract object archive benchmark summary");
|
||||
black_box(summary.file_count);
|
||||
});
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
group.finish();
|
||||
}
|
||||
|
||||
criterion_group!(
|
||||
benches,
|
||||
bench_tar_family_extract,
|
||||
bench_zip_helper_round_trip,
|
||||
bench_zip_helper_hotspot_breakdown,
|
||||
bench_zip_object_archive_extract
|
||||
);
|
||||
criterion_main!(benches);
|
||||
+1453
-773
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user