Add s3fs as an API of the filesystem benchmark suite

benchmarks/README.md

@@ -43,7 +43,9 @@ Locally and in CI, its virtual environment is `benchmarks/.venv`, separate from
 On the EC2 instance, the dedicated checkout's root `/opt/pyathena/.venv` holds the benchmark environment.
 The `polars1` and `polars2` dependency groups select Polars 1.x or 2.x, which PolarsCursor reads with different code paths.
 The environments above have `polars1`; a second environment beside each, `benchmarks/.venv-polars2` or `/opt/pyathena/.venv-polars2`, has `polars2`.
-The root `pyproject.toml` declares both groups and the root `dev` group as conflicting, so `uv.lock` resolves each of them separately.
+The `s3fs` dependency group installs [s3fs](https://github.com/fsspec/s3fs) for the filesystem suite's `s3fs` API in a third environment, `benchmarks/.venv-s3fs` or `/opt/pyathena/.venv-s3fs`, whose Polars version is not pinned.
+s3fs depends on aiobotocore, which pins botocore to a few patch releases, so the group is resolved apart from the other environments.
+The root `pyproject.toml` declares the three groups and the root `dev` group as conflicting, so `uv.lock` resolves each of them separately.
 The benchmark's default dependency groups include `polars1`; a sync with `--no-default-groups` needs `--group polars1` or `--group polars2`, or it installs whichever Polars version the lock resolves without them.
 The root `uv build -v` still builds only PyAthena, including a wheel built from its sdist.
 Benchmark dependencies do not become PyAthena runtime dependencies, and `uv sync --group dev` at the root does not install them.
@@ -55,7 +57,7 @@ just benchmark lint
 just benchmark test
 ```
 
-The recipes select `benchmarks/.venv` and Python 3.12, and sync that environment from the locked workspace; `just benchmark test` also runs the tests in `benchmarks/.venv-polars2`.
+The recipes select `benchmarks/.venv` and Python 3.12, and sync that environment from the locked workspace; `just benchmark test` also runs the tests in `benchmarks/.venv-polars2` and `benchmarks/.venv-s3fs`.
 The tests use local data, fake AWS clients, and child processes.
 They do not execute Athena queries or create AWS resources.
 `just benchmark format` formats the benchmark separately from the parent project.
@@ -222,28 +224,33 @@ It validates the full row count after construction, outside `init_seconds`.
 
 ### Filesystem suite
 
-The filesystem suite transfers one S3 object through PyAthena's `S3FileSystem` (API `sync`) and `AioS3FileSystem` (API `aio`) and does not query Athena.
+The filesystem suite transfers one S3 object through PyAthena's `S3FileSystem` (API `sync`) and `AioS3FileSystem` (API `aio`), and through s3fs's `S3FileSystem` (API `s3fs`), and does not query Athena.
 Its operations, selected with `--transport`, are:
 
-- `read`: one `read()` of the whole object from a file opened with `block_size`, which reads it as parallel range requests of that size and holds the parts and the joined object, about twice the object's size, in memory;
-- `write`: writes through a file opened with `block_size`, which uploads the object as a multipart upload with parts of that size;
-- `copy`: `cp_file()`, or `_cp_file()` of `AioS3FileSystem`, which copies an object over 5 GiB as a multipart copy with parts of `block_size`; the copies of smaller objects, single CopyObject requests, are recorded as unsupported; the copies exclude tags and annotations, which the generated objects do not have.
+- `read`: one `read()` of the whole object from a file opened with `block_size`; PyAthena's files read it as parallel range requests of that size and hold the parts and the joined object, about twice the object's size, in memory, while s3fs's file reads it as one stream through fsspec's read-ahead cache;
+- `get`: `get_file()` to a local file in the trial's temporary directory; PyAthena's reads one block at a time through a file opened with `block_size`, whose `bytes` cache requests the next block with the first read and one block at a time afterwards, and s3fs's reads parallel ranges of `chunksize` with `max_concurrency` requests at a time;
+- `write`: writes through a file opened with `block_size`, which uploads the object as a multipart upload with parts of that size; PyAthena's files upload up to `max_workers` parts at a time, and s3fs's file one part at a time;
+- `put`: `put_file()` from a sparse local file of the object's size, so that no disk read bounds the upload; PyAthena's uploads up to `max_workers` parts of `block_size` at a time, and s3fs's `max_concurrency` parts of `chunksize`, or one PutObject for an object below both twice `chunksize` and 5 GiB;
+- `copy`: `cp_file()`, or `_cp_file()` of `AioS3FileSystem`, with parts of `block_size`; PyAthena copies an object over 5 GiB as a multipart copy with up to `max_workers` part copies at a time, and s3fs copies one over 150 MiB (`s3fs.core.MANAGED_COPY_THRESHOLD`) as a multipart copy of one part at a time; the copies of smaller objects, single CopyObject requests, are recorded as unsupported; PyAthena's copies exclude tags and annotations, which the generated objects do not have.
 
-Each case sets the filesystem's `max_workers` and the S3 client's `max_pool_connections` through the connection's `s3_config`.
+Each case sets the filesystem's `max_workers` and the S3 client's `max_pool_connections`, through the connection's `s3_config` for PyAthena, and as `max_concurrency` and `config_kwargs` for s3fs, which creates its own client from the session's credentials.
 The `[filesystem]` section of `config.toml` sets the swept values, 8 and 40 workers and pools of 10 and 40 by default, and a `block_size` of 8 MiB.
 The suite's scales are the object sizes in bytes in `[filesystem] sizes`, not the row counts in `[scales]`: 64 MiB (small), 256 MiB (medium), 1 GiB (large), and 6 GiB (xlarge) by default.
+A `get` also writes the object to the trial's temporary directory, which the parent process measures as the peak temp dir and removes; on the Amazon Linux 2023 hosts, that directory is on the tmpfs `/tmp`, so the object stays in memory outside the process's RSS. A `put` reads a sparse file, which no disk bounds.
 
 The `sync` API calls the filesystem on the trial's thread, and a file's parallel requests run in an `S3ThreadPoolExecutor`.
-The `aio` API runs a read or write in `asyncio.to_thread()`, as `AioS3FSCursor` reads result files, and a file's parallel requests run through `S3AioExecutor`, dispatched on the filesystem's event loop, fsspec's IO thread, to a pool of `max_workers` threads of the file.
+The `aio` API runs a read, get, write, or put in `asyncio.to_thread()`, as `AioS3FSCursor` reads result files, and a file's parallel requests run through `S3AioExecutor`, dispatched on the filesystem's event loop, fsspec's IO thread, to a pool of `max_workers` threads of the file; its `get_file()` and `put_file()` run the synchronous filesystem's in the filesystem's thread pool.
 Its copy is a coroutine awaited on the trial's event loop, whose part copies run in the filesystem's pool of `max_workers` threads.
 Both event loops keep Python's default executor, which runs only the trial's `asyncio.to_thread()` call.
 Before #1159, the parallel requests of the `aio` API ran in the default executors, whose `min(32, os.cpu_count() + 4)` threads bound them.
+The `s3fs` API calls s3fs on the trial's thread; each call waits for a coroutine on fsspec's IO loop, where aiobotocore sends the requests, so the trial's event loop lag is not sampled.
+Its cases run in the `s3fs` environment: `jobs` marks them for it, without a suffix to their IDs, and rejects `--environment` with another environment for them; a `run` in an environment without s3fs records them as unsupported.
 
 The source objects, of zero bytes, are uploaded with boto3 below the run's scratch prefix and recorded under `objects` in the manifest.
 `prepare --filesystem-scale` uploads them with the manifest, so that every fleet host reads the same objects.
 Otherwise, `run` uploads the missing objects before its first trial, and a fleet worker uploads them on its host, to its own copy of the manifest.
 Writes and copies go to `trials/<trial>/object` below the scratch prefix.
-`total_seconds` and the event-loop lag cover the transfer; afterwards, the trial checks the size of the data read or of the written or copied object, and deletes the written or copied object.
+`total_seconds` and the event-loop lag cover the transfer; afterwards, the trial checks the size of the data read, of the downloaded file, or of the written, uploaded, or copied object, and deletes the written, uploaded, or copied object.
 `cleanup` removes the source objects and any remaining trial objects, the latter also with `--trials-only`.
 The summary shows the throughput in MiB per second and the peak thread count.
 A manifest for this suite alone needs no snapshot: without `--scale`, `prepare` creates none.
@@ -258,11 +265,11 @@ uv run --no-sync python -m pyathena_bench run \
 
 Reads and writes are bounded by the host's sustained network bandwidth: 3.75 Gbps on `r7g.2xlarge` and 25 Gbps on `c8gn.2xlarge`, both with 8 vCPUs, so that the default executor of an event loop has 12 threads on either.
 Copies are not: their UploadPartCopy requests name the source range, and S3 copies the bytes.
-`c8gn.2xlarge` has 16 GiB of memory, in which the 6 GiB read, holding about twice the object, leaves no margin below the default RSS limit; on it, select reads and writes up to large, where copies are unsupported, and writes and copies at xlarge:
+`c8gn.2xlarge` has 16 GiB of memory, in which the 6 GiB read, holding about twice the object, leaves no margin below the default RSS limit; on it, select reads, gets, writes, and puts up to large, where PyAthena's copies are unsupported, and writes and copies at xlarge, where a get would hold 6 GiB in the tmpfs:
 
 ```bash
 uv run --no-sync python -m pyathena_bench jobs \
-  --suite filesystem --scale small medium large --transport read write > results/jobs-filesystem.json
+  --suite filesystem --scale small medium large --transport read get write put > results/jobs-filesystem.json
 uv run --no-sync python -m pyathena_bench jobs \
   --suite filesystem --scale xlarge --transport write copy > results/jobs-filesystem-xlarge.json
 ```
@@ -293,7 +300,7 @@ Athena result reuse and library query caches are disabled.
 | Process high-water RSS | OS high-water value, including imports and process startup |
 | Event-loop lag | Delay beyond the heartbeat interval; 10 ms interval by default |
 | Thread metrics | Observed process threads and CPU time since the baseline, including native library threads |
-| Peak temp dir | Largest sampled size of the trial's Polars temporary directory |
+| Peak temp dir | Largest sampled size of the trial's temporary directory: Polars' file cache, and the local file of the filesystem suite's `get` and `put` |
 
 Some cursors eagerly read data during `execute()`, while others fetch lazily.
 Consequently, `consume_seconds` alone is not a fair retrieval comparison.
@@ -304,7 +311,7 @@ Client result time excludes the delay between server completion and its observat
 API row conversion and native DataFrame/Table access appear as separate cases.
 Sampled RSS can miss short peaks, and the constructor suite's RSS includes its subsequent validation read.
 If the operating system denies access to per-thread CPU times, `thread_cpu_available` is false; thread counts and RSS are still recorded.
-Each trial process receives its own `POLARS_TEMP_DIR`, which the parent samples with RSS and removes after the trial.
+Each trial process receives its own temporary directory, as `POLARS_TEMP_DIR` and in its payload, which the parent samples with RSS and removes after the trial.
 Polars before 1.39.0 downloaded the whole S3 object of a lazy CSV scan into a file cache in this directory before yielding batches and left the file after the process exited; PyAthena requires Polars 1.39.0 or later, whose scans do not write that cache.
 When the temporary directory is a tmpfs, as `/tmp` is on Amazon Linux 2023, files written there use memory that RSS does not include.
 
@@ -318,7 +325,7 @@ When the temporary directory is a tmpfs, as `/tmp` is on Amazon Linux 2023, file
 | SQLAlchemy (`sqla_*`) | `pandas.read_sql_query()` through an engine of each PyAthena driver, whole or with `chunksize` |
 | Wrangler CSV | Scalar projections, including the json and temporal workloads; nested cases are marked unsupported |
 | Wrangler CTAS and UNLOAD | Full DataFrame or chunks, including nested workloads |
-| S3 filesystem (`s3`) | Read, write, and multipart copy of one object with `S3FileSystem` and `AioS3FileSystem` |
+| S3 filesystem (`s3`) | Read, get, write, put, and multipart copy of one object with `S3FileSystem`, `AioS3FileSystem`, and s3fs |
 
 The SQLAlchemy families `sqla_rest`, `sqla_pandas`, `sqla_arrow`, `sqla_polars`, and `sqla_s3fs` read with the synchronous API in the single-query suite only, and the result-file drivers also read UNLOAD results.
 Their engines connect with the same workgroup, output location, polling interval, and disabled result reuse as the cursor cases, their statements run without the client cache (`cache_size` defaults to 0), and their cursors take the same cursor options; with `chunksize`, the pandas and Polars drivers' cursors also read the result file in chunks.

benchmarks/cloudformation/benchmark.yaml

@@ -363,6 +363,7 @@ Resources:
               uv python install ${PythonVersion}
               uv sync --directory /opt/pyathena/benchmarks --locked --no-default-groups --group polars1 --python ${PythonVersion}
               UV_PROJECT_ENVIRONMENT=/opt/pyathena/.venv-polars2 uv sync --directory /opt/pyathena/benchmarks --locked --no-default-groups --group polars2 --python ${PythonVersion}
+              UV_PROJECT_ENVIRONMENT=/opt/pyathena/.venv-s3fs uv sync --directory /opt/pyathena/benchmarks --locked --no-default-groups --group s3fs --python ${PythonVersion}
               chmod -R a+rX /opt/uv-python
               chown -R ec2-user:ec2-user /opt/pyathena
               printf '%s\n' '${AWS::StackId}' > /opt/pyathena/benchmarks/stack-id.txt

benchmarks/pyathena_bench/__main__.py

@@ -99,7 +99,7 @@ def parser() -> argparse.ArgumentParser:
         command.add_argument("--scale", nargs="+", required=True)
         command.add_argument("--family", choices=ALL_FAMILIES, nargs="+")
         command.add_argument(
-            "--api", choices=("sync", "thread", "aio", "direct", "to_thread"), nargs="+"
+            "--api", choices=("sync", "thread", "aio", "direct", "to_thread", "s3fs"), nargs="+"
         )
         command.add_argument(
             "--transport",

benchmarks/pyathena_bench/cases.py

@@ -28,9 +28,19 @@
     "sqla_s3fs": "s3fs",
 }
 # The filesystem suite transfers one S3 object with PyAthena's S3FileSystem (API "sync")
-# or AioS3FileSystem (API "aio"); its transport is the operation.
+# or AioS3FileSystem (API "aio"), or with s3fs (API "s3fs"); its transport is the
+# operation: whole reads and writes through an opened file, get_file() and put_file()
+# through a local file, and a copy.
 FILESYSTEM_FAMILY = "s3"
-FILESYSTEM_OPERATIONS = ("read", "write", "copy")
+FILESYSTEM_APIS = ("sync", "aio", "s3fs")
+FILESYSTEM_OPERATIONS = ("read", "get", "write", "put", "copy")
+# s3fs copies objects above its ``s3fs.core.MANAGED_COPY_THRESHOLD`` as multipart
+# copies, and PyAthena those above its largest part.
+SINGLE_COPY_THRESHOLDS = {
+    "sync": S3Core.MULTIPART_UPLOAD_MAX_PART_SIZE,
+    "aio": S3Core.MULTIPART_UPLOAD_MAX_PART_SIZE,
+    "s3fs": 150 * 2**20,
+}
 ALL_FAMILIES = (*FAMILIES, "wrangler", *SQLALCHEMY_FAMILIES, FILESYSTEM_FAMILY)
 SUITES = ("single", "concurrent", "init", "filesystem")
 SHAPES = ("flat", "nested", "json", "temporal")
@@ -71,9 +81,9 @@ def id(self) -> str:
     def unsupported_reason(self, settings: Settings, scale: str) -> str | None:
         """Return why the case does not run at a scale.
 
-        A copy of an object up to ``S3Core.MULTIPART_UPLOAD_MAX_PART_SIZE`` is
-        a single CopyObject request, which neither ``max_workers`` nor the
-        connection pool affects.
+        A copy of an object up to the API's threshold in
+        ``SINGLE_COPY_THRESHOLDS`` is a single CopyObject request, which
+        neither ``max_workers`` nor the connection pool affects.
 
         Args:
             settings: Configuration with the filesystem suite's object sizes.
@@ -85,9 +95,11 @@ def unsupported_reason(self, settings: Settings, scale: str) -> str | None:
         if (
             self.suite == "filesystem"
             and self.transport == "copy"
-            and settings.filesystem_sizes[scale] <= S3Core.MULTIPART_UPLOAD_MAX_PART_SIZE
+            and settings.filesystem_sizes[scale] <= SINGLE_COPY_THRESHOLDS[self.api]
         ):
-            return "Objects up to 5 GiB are copied with a single CopyObject request"
+            threshold = SINGLE_COPY_THRESHOLDS[self.api] // 2**20
+            unit = f"{threshold // 1024} GiB" if threshold % 1024 == 0 else f"{threshold} MiB"
+            return f"Objects up to {unit} are copied with a single CopyObject request"
         return self.unsupported
 
 
@@ -136,7 +148,7 @@ def matrix(settings: Settings, suite: str, shape: str) -> list[Case]:
                 pool_connections=pool,
             )
             for operation in FILESYSTEM_OPERATIONS
-            for api in ("sync", "aio")
+            for api in FILESYSTEM_APIS
             for workers in settings.filesystem_max_workers
             for pool in settings.filesystem_pool_connections
         ]