You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/14 20:11:45 UTC
(datafusion) branch gh-readonly-queue/main/pr-24093-eb71b22061e0fd1a075704732315a295eaeffa87 created (now 8c349a9e90)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-24093-eb71b22061e0fd1a075704732315a295eaeffa87
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 8c349a9e90 bench: model remote object storage faithfully for --simulate-latency (#24093)
This branch includes the following new commits:
new 8c349a9e90 bench: model remote object storage faithfully for --simulate-latency (#24093)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: bench: model remote object storage faithfully for --simulate-latency (#24093)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-24093-eb71b22061e0fd1a075704732315a295eaeffa87
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 8c349a9e90da5f097ccccff54aa6e58cacc1c14c
Author: Adrian Garcia Badaracco <17...@users.noreply.github.com>
AuthorDate: Wed Oct 14 20:11:37 2026 +0000
bench: model remote object storage faithfully for --simulate-latency (#24093)
## Which issue does this PR close?
- None.
## Rationale for this change
`--simulate-latency` must make benchmarks pay what S3 or GCS would
charge. The current `LatencyObjectStore` does not, for three reasons:
1. **It charges one request for a vectored read.** It overrides
`get_ranges` and sleeps once. `AmazonS3` and `GoogleCloudStorage` do not
override `get_ranges`. The trait default merges ranges that are less
than 1MiB apart, and sends each merged range as a separate GET. Thus a
Parquet scan that reads 2 distant column chunks costs 2 requests on S3,
but 1 request in the simulation. An IO change that alters the request
count is invisible to the benchmark.
2. **Bytes are free.** A 4 byte read and a 100MB read cost the same.
Thus merging ranges across a gap always looks free.
3. **A LIST costs one request, independent of the key count.** S3 and
GCS return 1000 keys per page, and each page needs the token of the
previous page. Thus a listing of 5000 keys costs 5 serial requests.
## What changes are included in this PR?
`SimulatedObjectStore` replaces `LatencyObjectStore`. Each GET costs:
1. one connection from a pool of 128,
2. a time to first byte, drawn at random from a measured distribution
(see below),
3. a transfer time at 90MiB/s per connection, charged while the caller
reads the body.
A HEAD costs one connection and one HEAD time to first byte. Each LIST
page costs one connection and one LIST time to first byte.
The store does not override `get_ranges`, so arrow-rs merges and fans
out the ranges exactly as it does against S3. The store overrides
`list_with_offset`, because S3 and GCS start that listing on the server.
Writes, copies and deletes go directly to the local filesystem without a
cost, because benchmarks do not measure them.
These are not modelled: request failures, retries, and a shared network
limit across connections.
### Calibration
I measured S3 and GCS on 2026-10-13 with the `object_store` clients,
from a VM in the same region as the bucket:
- S3: c7i.2xlarge in us-east-2, new bucket with 20k small keys and 64
objects of 128MiB.
- GCS: e2-standard-8 in europe-west1, existing bucket with real data,
200k keys listed.
| | S3 | GCS |
|---|---|---|
| GET time to first byte, P50 / P95 | 24 / 30 ms | 75 / 210 ms |
| HEAD, P50 / P95 | 11 / 16 ms | 18 / 28 ms |
| LIST page, P50 / P95 | 50 / 60 ms | 83 / 170 ms |
| One connection, 64MiB GET | ~100 MiB/s | ~80 MiB/s |
| Total, all connections | ~1.4 GiB/s at 32 connections | ~1.9 GiB/s at
128 connections |
S3 is approximately 3x faster to the first byte than GCS, thus one
distribution cannot fit both. Each latency table is an equal mix: 10
values at the 5%, 15%, ..., 95% quantiles of S3, and 10 values at the
same quantiles of GCS. Thus each request is S3-like or GCS-like with
equal probability. The throughput of one connection is 90MiB/s, the mean
of the two. The total limit is the network limit of the VM, not of the
store, thus it is not modelled.
## Are these changes tested?
Yes. 7 tests use tokio's paused clock, so the costs are exact and not
flaky:
| Test | What it shows |
|---|---|
| `distant_ranges_are_separate_requests` | Ranges more than 1MiB apart
are 2 requests. |
| `near_ranges_are_one_request` | Ranges less than 1MiB apart are 1
merged request. |
| `distant_column_chunks_cost_more_requests` | A real DataFusion Parquet
scan of 2 distant columns costs 1 more request than a scan of 1 column.
|
| `get_costs_ttfb_plus_transfer` | 256KiB at 1MiB/s costs TTFB + 250ms.
|
| `head_pays_only_the_head_ttfb` | A HEAD costs only the HEAD time to
first byte, not a transfer. |
| `list_costs_one_request_per_page` | 0, 1, 3, 4, 6 and 7 keys at 3 keys
per page cost 1, 1, 1, 2, 2 and 3 pages. |
| `list_with_offset_skips_earlier_pages` | A listing that starts at key
6 of 9 costs 1 page, not 3. |
### Benchmark results
The benchmark CI (`c4a-highmem-16`), with `SIMULATE_LATENCY=true` on
both sides. The baseline is `LatencyObjectStore` on main (45de64c).
Minimum time per query:
| Suite | Main | This PR | Change | Queries slower / faster / no change
|
|---|---|---|---|---|
| tpch | 16.9 s | 18.2 s | 1.08x slower | 14 / 6 / 2 |
| tpcds | 57.1 s | 47.9 s | 1.19x faster | 16 / 75 / 8 |
| clickbench_partitioned | 85.3 s | 126.0 s | 1.48x slower | 29 / 2 / 12
|
- **ClickBench is slower because bytes have a cost now.** Q23 (`SELECT
*` with `LIKE`) reads almost all columns and is 3.3x slower. Q36-Q42
(selective on one `CounterID`) send many small, scattered reads and are
1.6-3.3x slower, because each merged range is a separate request.
- **TPC-DS is faster because metadata is cheaper.** It reads many small
tables. HEAD now has its own, lower latency, and LIST pages are cheaper.
- **TPC-H is approximately the same.** The bandwidth cost of the large
tables and the lower metadata cost approximately cancel.
Results of `--simulate-latency` before this change are not comparable
with results after it.
## Are there any user-facing changes?
No. This change is in the benchmarks only. `--simulate-latency` and
`SIMULATE_LATENCY` keep their names.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5.5 <no...@anthropic.com>
---
benchmarks/Cargo.toml | 1 +
benchmarks/bench.sh | 3 +-
benchmarks/src/util/latency_object_store.rs | 163 --------
benchmarks/src/util/mod.rs | 2 +-
benchmarks/src/util/options.rs | 23 +-
benchmarks/src/util/simulated_object_store.rs | 530 ++++++++++++++++++++++++++
6 files changed, 548 insertions(+), 174 deletions(-)
diff --git a/benchmarks/Cargo.toml b/benchmarks/Cargo.toml
index cd222c3a5e..b5e19755b5 100644
--- a/benchmarks/Cargo.toml
+++ b/benchmarks/Cargo.toml
@@ -67,6 +67,7 @@ toml = "1.1.3"
[dev-dependencies]
datafusion-proto = { workspace = true, features = ["parquet"] }
tempfile = { workspace = true }
+tokio = { workspace = true, features = ["test-util"] }
[[bench]]
harness = false
diff --git a/benchmarks/bench.sh b/benchmarks/bench.sh
index 66b6dcd1b1..65ea19651e 100755
--- a/benchmarks/bench.sh
+++ b/benchmarks/bench.sh
@@ -188,7 +188,8 @@ CARGO_COMMAND command that runs the benchmark binary
DATAFUSION_DIR directory to use (default $DATAFUSION_DIR)
RESULTS_NAME folder where the benchmark files are stored
PREFER_HASH_JOIN Prefer hash join algorithm (default true)
-SIMULATE_LATENCY Simulate object store latency to mimic S3 (default false)
+SIMULATE_LATENCY Serve local files as if they were remote object storage: per-request
+ latency, per-connection bandwidth and paginated LIST (default false)
DATAFUSION_* Set the given datafusion configuration
"
exit 1
diff --git a/benchmarks/src/util/latency_object_store.rs b/benchmarks/src/util/latency_object_store.rs
deleted file mode 100644
index 9770fa3407..0000000000
--- a/benchmarks/src/util/latency_object_store.rs
+++ /dev/null
@@ -1,163 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you 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.
-
-//! An ObjectStore wrapper that adds simulated S3-like latency to get and list operations.
-//!
-//! Draws each request's latency at random from a fixed distribution inspired
-//! by real S3 performance:
-//! - P50: ~30ms
-//! - P75-P90: ~100-120ms
-//! - P99: ~150-200ms
-//!
-//! The draws are independent. A request's latency does not depend on how many
-//! requests came before it, so a change that adds or removes requests in one
-//! query does not change the latencies of the requests in the next query.
-
-use std::fmt;
-use std::time::Duration;
-
-use async_trait::async_trait;
-use futures::StreamExt;
-use futures::stream::BoxStream;
-use object_store::path::Path;
-use object_store::{
- CopyOptions, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta,
- ObjectStore, PutMultipartOptions, PutOptions, PutPayload, PutResult, Result,
-};
-use rand::Rng;
-
-/// GET latency distribution, inspired by S3 latencies. Each GET draws one
-/// value at random.
-/// 20 values: 11x P50 (~25-35ms), 5x P75-P90 (~70-110ms), 2x P95 (~120-150ms), 2x P99 (~180-200ms)
-/// Sorted: 25,25,28,28,30,30,30,30,32,32,35, 70,85,100,100,110, 130,150, 180,200
-/// P50≈32ms, P90≈110ms, P99≈200ms
-const GET_LATENCIES_MS: &[u64] = &[
- 30, 100, 25, 85, 32, 200, 28, 130, 35, 70, 30, 150, 30, 110, 28, 180, 32, 25, 100, 30,
-];
-
-/// LIST latency distribution, generally higher than GET.
-/// 20 values: 11x P50 (~40-70ms), 5x P75-P90 (~120-180ms), 2x P95 (~200-250ms), 2x P99 (~300-400ms)
-/// Sorted: 40,40,50,50,55,55,60,60,65,65,70, 120,140,160,160,180, 210,250, 300,400
-/// P50≈65ms, P90≈180ms, P99≈400ms
-const LIST_LATENCIES_MS: &[u64] = &[
- 55, 160, 40, 140, 65, 400, 50, 210, 70, 120, 60, 250, 55, 180, 50, 300, 65, 40, 160,
- 60,
-];
-
-/// An ObjectStore wrapper that injects simulated latency on get and list calls.
-#[derive(Debug)]
-pub struct LatencyObjectStore<T: ObjectStore> {
- inner: T,
-}
-
-impl<T: ObjectStore> LatencyObjectStore<T> {
- pub fn new(inner: T) -> Self {
- Self { inner }
- }
-
- fn next_get_latency(&self) -> Duration {
- sample(GET_LATENCIES_MS)
- }
-
- fn next_list_latency(&self) -> Duration {
- sample(LIST_LATENCIES_MS)
- }
-}
-
-/// One latency drawn at random from `distribution`.
-///
-/// Do not use a shared counter or a hash of the request instead. A counter
-/// makes a latency depend on the requests made before it. A hash gives a
-/// request shape the same latency in every iteration, so two builds that
-/// make differently shaped requests do not average out.
-fn sample(distribution: &[u64]) -> Duration {
- let idx = rand::rng().random_range(0..distribution.len());
- Duration::from_millis(distribution[idx])
-}
-
-impl<T: ObjectStore> fmt::Display for LatencyObjectStore<T> {
- fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
- write!(f, "LatencyObjectStore({})", self.inner)
- }
-}
-
-#[async_trait]
-impl<T: ObjectStore> ObjectStore for LatencyObjectStore<T> {
- async fn put_opts(
- &self,
- location: &Path,
- payload: PutPayload,
- opts: PutOptions,
- ) -> Result<PutResult> {
- self.inner.put_opts(location, payload, opts).await
- }
-
- async fn put_multipart_opts(
- &self,
- location: &Path,
- opts: PutMultipartOptions,
- ) -> Result<Box<dyn MultipartUpload>> {
- self.inner.put_multipart_opts(location, opts).await
- }
-
- async fn get_opts(&self, location: &Path, options: GetOptions) -> Result<GetResult> {
- tokio::time::sleep(self.next_get_latency()).await;
- self.inner.get_opts(location, options).await
- }
-
- async fn get_ranges(
- &self,
- location: &Path,
- ranges: &[std::ops::Range<u64>],
- ) -> Result<Vec<bytes::Bytes>> {
- tokio::time::sleep(self.next_get_latency()).await;
- self.inner.get_ranges(location, ranges).await
- }
-
- fn delete_stream(
- &self,
- locations: BoxStream<'static, Result<Path>>,
- ) -> BoxStream<'static, Result<Path>> {
- self.inner.delete_stream(locations)
- }
-
- fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, Result<ObjectMeta>> {
- let latency = self.next_list_latency();
- let stream = self.inner.list(prefix);
- futures::stream::once(async move {
- tokio::time::sleep(latency).await;
- futures::stream::empty()
- })
- .flatten()
- .chain(stream)
- .boxed()
- }
-
- async fn list_with_delimiter(&self, prefix: Option<&Path>) -> Result<ListResult> {
- tokio::time::sleep(self.next_list_latency()).await;
- self.inner.list_with_delimiter(prefix).await
- }
-
- async fn copy_opts(
- &self,
- from: &Path,
- to: &Path,
- options: CopyOptions,
- ) -> Result<()> {
- self.inner.copy_opts(from, to, options).await
- }
-}
diff --git a/benchmarks/src/util/mod.rs b/benchmarks/src/util/mod.rs
index 6dc11c0f42..b55ed49911 100644
--- a/benchmarks/src/util/mod.rs
+++ b/benchmarks/src/util/mod.rs
@@ -16,10 +16,10 @@
// under the License.
//! Shared benchmark utilities
-pub mod latency_object_store;
mod memory;
mod options;
mod run;
+pub mod simulated_object_store;
pub use memory::print_memory_stats;
pub use options::CommonOpt;
diff --git a/benchmarks/src/util/options.rs b/benchmarks/src/util/options.rs
index 4a1c14674a..3d5906f4f3 100644
--- a/benchmarks/src/util/options.rs
+++ b/benchmarks/src/util/options.rs
@@ -33,7 +33,9 @@ use datafusion::{
use datafusion_common::{DataFusionError, Result};
use object_store::local::LocalFileSystem;
-use super::latency_object_store::LatencyObjectStore;
+use super::simulated_object_store::{
+ BYTES_PER_SECOND, MAX_CONNECTIONS, SimulatedObjectStore,
+};
// Common benchmark options (don't use doc comments otherwise this doc
// shows up in help files)
@@ -69,8 +71,10 @@ pub struct CommonOpt {
#[arg(short, long, env)]
pub debug: bool,
- /// Simulate object store latency to mimic remote storage (e.g. S3).
- /// Adds random latency in the range 20-200ms to each object store operation.
+ /// Serve the local filesystem as if it were remote object storage (e.g. S3
+ /// or GCS): per-request latency, per-connection bandwidth, a bounded
+ /// connection pool, and paginated LIST. See
+ /// [`crate::util::simulated_object_store`].
#[arg(long = "simulate-latency", env)]
pub simulate_latency: bool,
}
@@ -138,17 +142,18 @@ impl CommonOpt {
Ok(rt_builder)
}
- /// Build the runtime environment, optionally wrapping the local filesystem
- /// with a throttled object store to simulate remote storage latency.
+ /// Build the runtime environment, optionally presenting the local
+ /// filesystem as remote object storage.
pub fn build_runtime(&self) -> Result<Arc<RuntimeEnv>> {
let rt = self.runtime_env_builder()?.build_arc()?;
if self.simulate_latency {
- let store: Arc<dyn object_store::ObjectStore> =
- Arc::new(LatencyObjectStore::new(LocalFileSystem::new()));
+ let store = SimulatedObjectStore::new(LocalFileSystem::new());
let url = ObjectStoreUrl::parse("file:///")?;
- rt.register_object_store(url.as_ref(), store);
+ rt.register_object_store(url.as_ref(), Arc::new(store));
println!(
- "Simulating S3-like object store latency (get: 25-200ms, list: 40-400ms)"
+ "Simulating remote object storage (get TTFB: 19-230ms, head TTFB: \
+ 8-28ms, list TTFB: 40-200ms/page, {} MB/s per connection, {MAX_CONNECTIONS} connections)",
+ BYTES_PER_SECOND / (1024 * 1024),
);
}
Ok(rt)
diff --git a/benchmarks/src/util/simulated_object_store.rs b/benchmarks/src/util/simulated_object_store.rs
new file mode 100644
index 0000000000..a24bd2af77
--- /dev/null
+++ b/benchmarks/src/util/simulated_object_store.rs
@@ -0,0 +1,530 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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.
+
+//! An [`ObjectStore`] that makes a local filesystem behave like S3 or GCS.
+//!
+//! Each request costs:
+//!
+//! 1. a connection from a pool of [`MAX_CONNECTIONS`],
+//! 2. a time to first byte, drawn at random from a latency distribution,
+//! 3. a transfer time at [`BYTES_PER_SECOND`] for each response body.
+//!
+//! The store does not override [`ObjectStore::get_ranges`]. `AmazonS3` and
+//! `GoogleCloudStorage` do not override it either. The trait default merges
+//! ranges that are less than 1MiB apart and sends each merged range as a
+//! separate GET. Thus the simulated request count is the same as on S3. A
+//! wrapper that overrides `get_ranges` charges one request where S3 charges
+//! several.
+//!
+//! The store does not model request failures, retries or a shared network
+//! limit.
+
+use std::fmt;
+use std::sync::Arc;
+use std::time::Duration;
+
+use async_trait::async_trait;
+use futures::StreamExt;
+use futures::stream::BoxStream;
+use object_store::path::Path;
+use object_store::{
+ CopyOptions, GetOptions, GetResult, GetResultPayload, ListResult, MultipartUpload,
+ ObjectMeta, ObjectStore, PutMultipartOptions, PutOptions, PutPayload, PutResult,
+ Result,
+};
+use rand::Rng;
+use tokio::sync::{OwnedSemaphorePermit, Semaphore};
+use tokio::time::Instant;
+
+// The latency tables are an equal mix of S3 and GCS. Each table has 10 values
+// from S3 and 10 values from GCS, at the 5%, 15%, ..., 95% quantiles of each.
+// Thus one draw is S3-like or GCS-like with equal probability.
+//
+// Measured on 2026-10-13 with the `object_store` clients, from a VM in the
+// same region as the bucket:
+// - S3: c7i.2xlarge in us-east-2. GCS: e2-standard-8 in europe-west1.
+// - S3 is approximately 3x faster to the first byte than GCS.
+
+/// Time to first byte of a range GET.
+/// S3: P50≈24ms, P95≈30ms. GCS: P50≈75ms, P95≈210ms.
+const GET_TTFB_MS: &[u64] = &[
+ 19, 21, 22, 23, 24, 25, 26, 27, 29, 35, // S3
+ 30, 45, 55, 65, 75, 90, 110, 130, 165, 230, // GCS
+];
+
+/// Time to first byte of a HEAD.
+/// S3: P50≈11ms, P95≈16ms. GCS: P50≈18ms, P95≈28ms.
+const HEAD_TTFB_MS: &[u64] = &[
+ 8, 9, 10, 10, 11, 11, 12, 12, 13, 16, // S3
+ 13, 15, 16, 17, 18, 18, 19, 21, 23, 28, // GCS
+];
+
+/// Time to first byte of one LIST page.
+/// S3: P50≈50ms, P95≈60ms. GCS: P50≈83ms, P95≈170ms.
+const LIST_TTFB_MS: &[u64] = &[
+ 40, 45, 47, 48, 49, 50, 51, 52, 54, 60, // S3
+ 50, 62, 70, 75, 80, 86, 95, 106, 130, 200, // GCS
+];
+
+/// Throughput of one connection. For a 64MiB range GET, the measured median
+/// after the time to first byte is approximately 100MiB/s on S3 and 80MiB/s
+/// on GCS.
+pub const BYTES_PER_SECOND: u64 = 90 * 1024 * 1024;
+
+/// Maximum number of requests in flight.
+pub const MAX_CONNECTIONS: usize = 128;
+
+/// Maximum keys in one LIST page. S3 and GCS both use 1000.
+const LIST_PAGE_SIZE: usize = 1000;
+
+#[derive(Debug, Clone, Copy)]
+struct Costs {
+ get_ttfb_ms: &'static [u64],
+ head_ttfb_ms: &'static [u64],
+ list_ttfb_ms: &'static [u64],
+ bytes_per_second: u64,
+ list_page_size: usize,
+}
+
+/// Presents `T` (usually a `LocalFileSystem`) as remote object storage. See
+/// the [module docs](self).
+#[derive(Debug)]
+pub struct SimulatedObjectStore<T: ObjectStore> {
+ inner: T,
+ costs: Costs,
+ connections: Arc<Semaphore>,
+}
+
+impl<T: ObjectStore> SimulatedObjectStore<T> {
+ pub fn new(inner: T) -> Self {
+ let costs = Costs {
+ get_ttfb_ms: GET_TTFB_MS,
+ head_ttfb_ms: HEAD_TTFB_MS,
+ list_ttfb_ms: LIST_TTFB_MS,
+ bytes_per_second: BYTES_PER_SECOND,
+ list_page_size: LIST_PAGE_SIZE,
+ };
+ Self {
+ inner,
+ costs,
+ connections: Arc::new(Semaphore::new(MAX_CONNECTIONS)),
+ }
+ }
+
+ /// Charges one LIST request for each page of `keys`.
+ fn paged(
+ &self,
+ keys: BoxStream<'static, Result<ObjectMeta>>,
+ ) -> BoxStream<'static, Result<ObjectMeta>> {
+ let costs = self.costs;
+ let connections = Arc::clone(&self.connections);
+ futures::stream::unfold((keys, 0usize), move |(mut keys, seen)| {
+ let connections = Arc::clone(&connections);
+ async move {
+ let key = keys.next().await;
+ // Each page needs the token from the previous page, thus pages
+ // are serial. An empty listing also costs one request.
+ let new_page = seen % costs.list_page_size == 0;
+ if new_page && (key.is_some() || seen == 0) {
+ let _permit = round_trip(&connections, costs.list_ttfb_ms).await;
+ }
+ Some((key?, (keys, seen + 1)))
+ }
+ })
+ .boxed()
+ }
+}
+
+/// Waits for a free connection and for the time to first byte. The request
+/// keeps the connection until it drops the returned permit.
+async fn round_trip(
+ connections: &Arc<Semaphore>,
+ ttfb_ms: &[u64],
+) -> OwnedSemaphorePermit {
+ let permit = Arc::clone(connections)
+ .acquire_owned()
+ .await
+ .expect("connection pool is never closed");
+ let idx = rand::rng().random_range(0..ttfb_ms.len());
+ tokio::time::sleep(Duration::from_millis(ttfb_ms[idx])).await;
+ permit
+}
+
+fn transfer_time(bytes: u64, bytes_per_second: u64) -> Duration {
+ Duration::from_secs_f64(bytes as f64 / bytes_per_second as f64)
+}
+
+/// Charges transfer time while the caller reads the body. The connection stays
+/// in use until the body is fully read or dropped.
+fn throttle(
+ result: GetResult,
+ permit: OwnedSemaphorePermit,
+ bytes_per_second: u64,
+) -> GetResult {
+ let meta = result.meta.clone();
+ let range = result.range.clone();
+ let attributes = result.attributes.clone();
+ let extensions = result.extensions.clone();
+ // The deadline is absolute, so time spent on the local read and timer
+ // overshoot do not add to the simulated transfer time. The clock starts
+ // when the caller first reads the body.
+ let body = futures::stream::unfold(
+ (result.into_stream(), None, 0u64, permit),
+ move |(mut body, start, mut sent, permit)| async move {
+ let start = start.unwrap_or_else(Instant::now);
+ let chunk = body.next().await?;
+ if let Ok(bytes) = &chunk {
+ sent += bytes.len() as u64;
+ tokio::time::sleep_until(start + transfer_time(sent, bytes_per_second))
+ .await;
+ }
+ Some((chunk, (body, Some(start), sent, permit)))
+ },
+ );
+ GetResult {
+ payload: GetResultPayload::Stream(body.boxed()),
+ meta,
+ range,
+ attributes,
+ extensions,
+ }
+}
+
+impl<T: ObjectStore> fmt::Display for SimulatedObjectStore<T> {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ write!(f, "SimulatedObjectStore({})", self.inner)
+ }
+}
+
+/// Only reads and lists are charged. Writes go directly to `inner`, because
+/// benchmarks do not measure them.
+#[async_trait]
+impl<T: ObjectStore> ObjectStore for SimulatedObjectStore<T> {
+ async fn put_opts(
+ &self,
+ location: &Path,
+ payload: PutPayload,
+ opts: PutOptions,
+ ) -> Result<PutResult> {
+ self.inner.put_opts(location, payload, opts).await
+ }
+
+ async fn put_multipart_opts(
+ &self,
+ location: &Path,
+ opts: PutMultipartOptions,
+ ) -> Result<Box<dyn MultipartUpload>> {
+ self.inner.put_multipart_opts(location, opts).await
+ }
+
+ async fn get_opts(&self, location: &Path, options: GetOptions) -> Result<GetResult> {
+ let ttfb_ms = match options.head {
+ true => self.costs.head_ttfb_ms,
+ false => self.costs.get_ttfb_ms,
+ };
+ let permit = round_trip(&self.connections, ttfb_ms).await;
+ let result = self.inner.get_opts(location, options).await?;
+ // A HEAD result has a range that covers the full object, but the
+ // caller never reads its body, so it pays no transfer time.
+ Ok(throttle(result, permit, self.costs.bytes_per_second))
+ }
+
+ fn delete_stream(
+ &self,
+ locations: BoxStream<'static, Result<Path>>,
+ ) -> BoxStream<'static, Result<Path>> {
+ self.inner.delete_stream(locations)
+ }
+
+ fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, Result<ObjectMeta>> {
+ self.paged(self.inner.list(prefix))
+ }
+
+ // S3 and GCS override this to start the listing at `offset` on the server.
+ // The trait default lists all keys and then filters them.
+ fn list_with_offset(
+ &self,
+ prefix: Option<&Path>,
+ offset: &Path,
+ ) -> BoxStream<'static, Result<ObjectMeta>> {
+ self.paged(self.inner.list_with_offset(prefix, offset))
+ }
+
+ async fn list_with_delimiter(&self, prefix: Option<&Path>) -> Result<ListResult> {
+ let result = self.inner.list_with_delimiter(prefix).await?;
+ let entries = result.objects.len() + result.common_prefixes.len();
+ let pages = entries.div_ceil(self.costs.list_page_size).max(1);
+ for _ in 0..pages {
+ let _permit = round_trip(&self.connections, self.costs.list_ttfb_ms).await;
+ }
+ Ok(result)
+ }
+
+ async fn copy_opts(
+ &self,
+ from: &Path,
+ to: &Path,
+ options: CopyOptions,
+ ) -> Result<()> {
+ self.inner.copy_opts(from, to, options).await
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use std::sync::Mutex;
+
+ use bytes::Bytes;
+ use object_store::memory::InMemory;
+ use object_store::{ObjectStoreExt, PutPayload};
+
+ use super::*;
+
+ /// Records the range of each GET that reaches the inner store. On S3, each
+ /// one is a separate HTTP request.
+ #[derive(Debug, Default)]
+ struct Recording {
+ inner: InMemory,
+ gets: Mutex<Vec<std::ops::Range<u64>>>,
+ }
+
+ impl fmt::Display for Recording {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ write!(f, "Recording")
+ }
+ }
+
+ #[async_trait]
+ impl ObjectStore for Recording {
+ async fn put_opts(
+ &self,
+ location: &Path,
+ payload: PutPayload,
+ opts: PutOptions,
+ ) -> Result<PutResult> {
+ self.inner.put_opts(location, payload, opts).await
+ }
+
+ async fn put_multipart_opts(
+ &self,
+ location: &Path,
+ opts: PutMultipartOptions,
+ ) -> Result<Box<dyn MultipartUpload>> {
+ self.inner.put_multipart_opts(location, opts).await
+ }
+
+ async fn get_opts(
+ &self,
+ location: &Path,
+ options: GetOptions,
+ ) -> Result<GetResult> {
+ let result = self.inner.get_opts(location, options).await?;
+ self.gets.lock().unwrap().push(result.range.clone());
+ Ok(result)
+ }
+
+ fn delete_stream(
+ &self,
+ locations: BoxStream<'static, Result<Path>>,
+ ) -> BoxStream<'static, Result<Path>> {
+ self.inner.delete_stream(locations)
+ }
+
+ fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, Result<ObjectMeta>> {
+ self.inner.list(prefix)
+ }
+
+ async fn list_with_delimiter(&self, prefix: Option<&Path>) -> Result<ListResult> {
+ self.inner.list_with_delimiter(prefix).await
+ }
+
+ async fn copy_opts(
+ &self,
+ from: &Path,
+ to: &Path,
+ options: CopyOptions,
+ ) -> Result<()> {
+ self.inner.copy_opts(from, to, options).await
+ }
+ }
+
+ const TTFB_MS: u64 = 10;
+ const HEAD_TTFB_MS: u64 = 3;
+
+ /// A store with constant times to first byte, 1MiB/s and 3 keys per page.
+ async fn store(objects: &[(&str, usize)]) -> SimulatedObjectStore<Recording> {
+ let inner = Recording::default();
+ for (name, len) in objects {
+ let payload = PutPayload::from(vec![7u8; *len]);
+ inner.put(&Path::from(*name), payload).await.unwrap();
+ }
+ SimulatedObjectStore {
+ inner,
+ costs: Costs {
+ get_ttfb_ms: &[TTFB_MS],
+ head_ttfb_ms: &[HEAD_TTFB_MS],
+ list_ttfb_ms: &[TTFB_MS],
+ bytes_per_second: 1024 * 1024,
+ list_page_size: 3,
+ },
+ connections: Arc::new(Semaphore::new(MAX_CONNECTIONS)),
+ }
+ }
+
+ fn ms(elapsed: Duration) -> u64 {
+ elapsed.as_millis() as u64
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn distant_ranges_are_separate_requests() {
+ let store = store(&[("f", 8 << 20)]).await;
+ let ranges = [0..1024, (3 << 20)..(3 << 20) + 1024];
+
+ let data = store.get_ranges(&Path::from("f"), &ranges).await.unwrap();
+
+ assert_eq!(data, vec![Bytes::from(vec![7u8; 1024]); 2]);
+ assert_eq!(store.inner.gets.lock().unwrap().len(), 2);
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn near_ranges_are_one_request() {
+ let store = store(&[("f", 8 << 20)]).await;
+
+ store
+ .get_ranges(&Path::from("f"), &[0..1024, 2048..4096])
+ .await
+ .unwrap();
+
+ assert_eq!(*store.inner.gets.lock().unwrap(), vec![0..4096]);
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn get_costs_ttfb_plus_transfer() {
+ let store = store(&[("f", 1 << 20)]).await;
+
+ let start = Instant::now();
+ store
+ .get_range(&Path::from("f"), 0..256 * 1024)
+ .await
+ .unwrap();
+
+ // 10ms TTFB plus 256KiB at 1MiB/s.
+ assert_eq!(ms(start.elapsed()), TTFB_MS + 250);
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn head_pays_only_the_head_ttfb() {
+ let store = store(&[("f", 1 << 20)]).await;
+
+ let start = Instant::now();
+ store.head(&Path::from("f")).await.unwrap();
+
+ assert_eq!(ms(start.elapsed()), HEAD_TTFB_MS);
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn list_costs_one_request_per_page() {
+ for (keys, pages) in [(0, 1), (1, 1), (3, 1), (4, 2), (6, 2), (7, 3)] {
+ let names: Vec<_> = (0..keys).map(|i| format!("k{i}")).collect();
+ let objects: Vec<_> = names.iter().map(|n| (n.as_str(), 1)).collect();
+ let store = store(&objects).await;
+
+ let start = Instant::now();
+ let listed: Vec<_> = store.list(None).collect().await;
+
+ assert_eq!(listed.len(), keys);
+ assert_eq!(ms(start.elapsed()), pages * TTFB_MS, "{keys} keys");
+ }
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn list_with_offset_skips_earlier_pages() {
+ let names: Vec<_> = (0..9).map(|i| format!("k{i}")).collect();
+ let objects: Vec<_> = names.iter().map(|n| (n.as_str(), 1)).collect();
+ let store = store(&objects).await;
+
+ let start = Instant::now();
+ let listed: Vec<_> = store
+ .list_with_offset(None, &Path::from("k5"))
+ .collect()
+ .await;
+
+ // k6, k7 and k8 fit in one page.
+ assert_eq!(listed.len(), 3);
+ assert_eq!(ms(start.elapsed()), TTFB_MS);
+ }
+
+ /// A Parquet file with 8 columns. Random `i64` values do not compress, so
+ /// each column chunk is larger than 1MiB and is not merged with the next.
+ fn wide_parquet() -> Bytes {
+ use arrow::array::{ArrayRef, Int64Array};
+ use arrow::datatypes::{DataType, Field, Schema};
+ use arrow::record_batch::RecordBatch;
+ use parquet::arrow::ArrowWriter;
+ use rand::{SeedableRng, rngs::StdRng};
+
+ let mut rng = StdRng::seed_from_u64(42);
+ let fields: Vec<_> = (0..8)
+ .map(|i| Field::new(format!("c{i}"), DataType::Int64, false))
+ .collect();
+ let schema = Arc::new(Schema::new(fields));
+ let arrays: Vec<ArrayRef> = (0..8)
+ .map(|_| {
+ let values: Vec<i64> = (0..200_000).map(|_| rng.random()).collect();
+ Arc::new(Int64Array::from(values)) as ArrayRef
+ })
+ .collect();
+ let batch = RecordBatch::try_new(Arc::clone(&schema), arrays).unwrap();
+ let mut buf = Vec::new();
+ let mut writer = ArrowWriter::try_new(&mut buf, schema, None).unwrap();
+ writer.write(&batch).unwrap();
+ writer.close().unwrap();
+ Bytes::from(buf)
+ }
+
+ /// Returns the number of GET requests that the scan in `sql` sends.
+ async fn gets_for_query(sql: &str) -> usize {
+ use datafusion::execution::object_store::ObjectStoreUrl;
+ use datafusion::prelude::{ParquetReadOptions, SessionContext};
+
+ let store = Arc::new(store(&[]).await);
+ store
+ .put(&Path::from("t.parquet"), wide_parquet().into())
+ .await
+ .unwrap();
+ let ctx = SessionContext::new();
+ let url = ObjectStoreUrl::parse("mem://").unwrap();
+ ctx.register_object_store(url.as_ref(), Arc::clone(&store) as _);
+ ctx.register_parquet("t", "mem:///t.parquet", ParquetReadOptions::default())
+ .await
+ .unwrap();
+ // Do not count the reads of schema inference.
+ store.inner.gets.lock().unwrap().clear();
+ ctx.sql(sql).await.unwrap().collect().await.unwrap();
+ store.inner.gets.lock().unwrap().len()
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn distant_column_chunks_cost_more_requests() {
+ // DataFusion reads both column chunks with one `get_ranges` call. On
+ // S3 that is two requests, because the chunks are far apart.
+ let one = gets_for_query("SELECT sum(c0) FROM t").await;
+ let two = gets_for_query("SELECT sum(c0), sum(c7) FROM t").await;
+
+ assert_eq!(two, one + 1);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org