diff --git a/.github/workflows/python-testing.yml b/.github/workflows/python-testing.yml index 23061d612b..9f025647ad 100644 --- a/.github/workflows/python-testing.yml +++ b/.github/workflows/python-testing.yml @@ -41,9 +41,10 @@ on: jobs: rust-check: - # Fast gate: type-check both with and without CUDA stubs so duplicate - # stub definitions or cfg mismatches fail in ~30s instead of surfacing - # during the slower maturin build below. + # Early gate: type-check both with and without CUDA stubs so duplicate + # stub definitions or cfg mismatches fail quickly instead of surfacing + # during the slower maturin build below. Also runs qdp-core f32 Parquet + # CPU smoke tests (issue #1342). runs-on: ubuntu-latest steps: - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6 @@ -57,6 +58,17 @@ jobs: QDP_NO_CUDA: "1" run: cargo check --workspace --tests + # CPU smoke for the f32 Parquet pipeline (issue #1342). No GPU in CI, so + # GPU fidelity cases self-skip when the probe finds no functional GPU + # (stub build: kernel launch returns Err; no driver: cudarc panics and + # catch_unwind treats it as skip). Reader-level assertions still run. + # Scoped to these two binaries — other gpu_*.rs tests lack probe-skip. + - name: Cargo test (f32 Parquet CPU smoke) + working-directory: qdp + env: + QDP_NO_CUDA: "1" + run: cargo test -p qdp-core --test parquet_f32 --test parquet_f32_fidelity + test: needs: rust-check runs-on: ubuntu-latest diff --git a/qdp/qdp-core/tests/common/mod.rs b/qdp/qdp-core/tests/common/mod.rs index 7114ad613a..241012d0f7 100644 --- a/qdp/qdp-core/tests/common/mod.rs +++ b/qdp/qdp-core/tests/common/mod.rs @@ -17,7 +17,9 @@ #[cfg(target_os = "linux")] use std::sync::Arc; -use arrow::array::{FixedSizeListArray, Float64Array}; +use arrow::array::{ + Array, FixedSizeListArray, Float32Builder, Float64Array, Float64Builder, ListBuilder, +}; use arrow::datatypes::{DataType, Field, Schema}; use arrow::record_batch::RecordBatch; use parquet::arrow::ArrowWriter; @@ -81,6 +83,79 @@ pub fn write_fixed_size_list_parquet(path: &str, data: &[f64], sample_size: usiz writer.close().unwrap(); } +/// Writes a `List` Parquet file; each `sample_size` consecutive values in +/// `data` form one row. Mirrors [`write_list_parquet_f64`] for the f32 column path +/// (issue #1342 Parquet f32 fidelity tests). +#[allow(dead_code)] +#[allow(clippy::manual_is_multiple_of)] +pub fn write_list_parquet_f32(path: &str, data: &[f32], sample_size: usize) { + assert!(sample_size > 0, "sample_size must be > 0"); + assert!( + data.len() % sample_size == 0, + "Data length ({}) must be a multiple of sample size ({})", + data.len(), + sample_size + ); + use std::fs::File; + use std::sync::Arc; + + let mut builder = ListBuilder::new(Float32Builder::new()); + for row in data.chunks(sample_size) { + builder.values().append_slice(row); + builder.append(true); + } + let list_array = builder.finish(); + + let schema = Arc::new(Schema::new(vec![Field::new( + "data", + list_array.data_type().clone(), + true, + )])); + let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(list_array) as _]).unwrap(); + + let file = File::create(path).unwrap(); + let props = WriterProperties::builder().build(); + let mut writer = ArrowWriter::try_new(file, schema, Some(props)).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); +} + +/// Writes a `List` Parquet file; each `sample_size` consecutive values in +/// `data` form one row. Companion to [`write_list_parquet_f32`]. +#[allow(dead_code)] +#[allow(clippy::manual_is_multiple_of)] +pub fn write_list_parquet_f64(path: &str, data: &[f64], sample_size: usize) { + assert!(sample_size > 0, "sample_size must be > 0"); + assert!( + data.len() % sample_size == 0, + "Data length ({}) must be a multiple of sample size ({})", + data.len(), + sample_size + ); + use std::fs::File; + use std::sync::Arc; + + let mut builder = ListBuilder::new(Float64Builder::new()); + for row in data.chunks(sample_size) { + builder.values().append_slice(row); + builder.append(true); + } + let list_array = builder.finish(); + + let schema = Arc::new(Schema::new(vec![Field::new( + "data", + list_array.data_type().clone(), + true, + )])); + let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(list_array) as _]).unwrap(); + + let file = File::create(path).unwrap(); + let props = WriterProperties::builder().build(); + let mut writer = ArrowWriter::try_new(file, schema, Some(props)).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); +} + /// Returns a CUDA device handle, or `None` when CUDA is unavailable for the test environment. #[cfg(target_os = "linux")] #[allow(dead_code)] diff --git a/qdp/qdp-core/tests/parquet_f32_fidelity.rs b/qdp/qdp-core/tests/parquet_f32_fidelity.rs new file mode 100644 index 0000000000..58f40b9bdc --- /dev/null +++ b/qdp/qdp-core/tests/parquet_f32_fidelity.rs @@ -0,0 +1,257 @@ +// +// 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. + +//! End-to-end f32 Parquet fidelity tests — issue #1342. +//! +//! These tests extend `gpu_fidelity.rs` (which compares f32 vs f64 encodings on +//! in-memory data) by sourcing the data from Parquet: the same deterministic +//! amplitude data is written as both a `List` and a `List` +//! column, read back through `ParquetReader::` / `ParquetReader::`, and +//! encoded with the f32 / f64 kernels respectively. We then compare the GPU +//! state vectors with `fidelity_cross_precision`. +//! +//! Two layers: +//! * CPU smoke (no GPU, runs everywhere): the f32 reader and the f64 reader +//! return the same logical values up to the f64→f32 rounding. This is the +//! part that actually asserts in CI (ubuntu, no GPU). +//! * GPU fidelity (Linux + CUDA, skipped otherwise): full reader→kernel +//! pipeline. Thresholds match `gpu_fidelity.rs` — amplitude f32 norm +//! precision means ~1e-3 at 8–16 qubits, NOT 0.99999. The issue's +//! "≥ 0.99999 where applicable" does not apply to amplitude encoding; see +//! `gpu_fidelity.rs::compare_f32_f64_amplitude` for the established baseline. + +use std::sync::atomic::{AtomicUsize, Ordering}; + +use qdp_core::reader::{DataReader, NullHandling}; +use qdp_core::readers::parquet::ParquetReader; + +mod common; + +static FILE_COUNTER: AtomicUsize = AtomicUsize::new(0); + +/// Deterministic amplitude data, identical generator to `gpu_fidelity.rs`. +fn make_amplitude_data(num_samples: usize, sample_size: usize) -> Vec { + let total = num_samples * sample_size; + (0..total) + .map(|i| ((i as f64) + 1.0) / total as f64) + .collect() +} + +/// Returns `(f32_path, f64_path)` for two temp Parquet files holding the same +/// logical data as `List` and `List` respectively. +fn write_pair(data_f64: &[f64], sample_size: usize) -> (TempFile, TempFile) { + let n = FILE_COUNTER.fetch_add(1, Ordering::Relaxed); + let pid = std::process::id(); + let f32_path = std::env::temp_dir().join(format!("mahout_f32fid_{pid}_{n}_f32.parquet")); + let f64_path = std::env::temp_dir().join(format!("mahout_f32fid_{pid}_{n}_f64.parquet")); + + let data_f32: Vec = data_f64.iter().map(|&x| x as f32).collect(); + common::write_list_parquet_f32(f32_path.to_str().unwrap(), &data_f32, sample_size); + common::write_list_parquet_f64(f64_path.to_str().unwrap(), data_f64, sample_size); + + (TempFile(f32_path), TempFile(f64_path)) +} + +struct TempFile(std::path::PathBuf); + +impl Drop for TempFile { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.0); + } +} + +// --------------------------------------------------------------------------- +// CPU smoke (no GPU) — this is what actually asserts in CI. +// --------------------------------------------------------------------------- + +/// The f32 reader and the f64 reader return the same shape, and the f32 values +/// equal the f64 values narrowed with `as f32` (Arrow cast == IEEE +/// round-to-nearest, same as the f64→f32 path the kernel would see). +#[test] +fn test_parquet_f32_f64_reader_consistency() { + let num_samples = 4; + let sample_size = 256; // 8 qubits + let data_f64 = make_amplitude_data(num_samples, sample_size); + let (f32_file, f64_file) = write_pair(&data_f64, sample_size); + + let mut reader_f32 = + ParquetReader::::new(&f32_file.0, None, NullHandling::FillZero).unwrap(); + let (data32, n32, s32) = reader_f32.read_batch().unwrap(); + + let mut reader_f64 = + ParquetReader::::new(&f64_file.0, None, NullHandling::FillZero).unwrap(); + let (data64, n64, s64) = reader_f64.read_batch().unwrap(); + + assert_eq!((n32, s32), (num_samples, sample_size)); + assert_eq!((n64, s64), (num_samples, sample_size)); + assert_eq!(data32.len(), data64.len()); + + for (i, (&v32, &v64)) in data32.iter().zip(data64.iter()).enumerate() { + assert_eq!( + v32, v64 as f32, + "f32 reader value {v32} != f64 narrowed value {} at index {i}", + v64 as f32 + ); + } +} + +// --------------------------------------------------------------------------- +// GPU fidelity (Linux + CUDA) — mirrors compare_f32_f64_amplitude, Parquet src. +// --------------------------------------------------------------------------- + +#[cfg(target_os = "linux")] +mod gpu { + use super::*; + use qdp_core::Precision; + use qdp_core::gpu::metrics::{ + download_complex_f32, download_complex_f64, fidelity_cross_precision, + }; + + /// Returns `true` only when CUDA kernels actually launch. Two distinct + /// no-GPU environments must both be detected as "skip": + /// * No toolkit but a driver present (the `QDP_NO_CUDA` stub): engine and + /// device creation and host→device copies all succeed — only a kernel + /// launch fails with code 999, surfacing as an `Err` here. + /// * No driver at all (CI runner without `libcuda.so`): `cudarc` loads the + /// driver dynamically and **panics** on first use rather than returning + /// an `Err`, so `.ok()` does not catch it. + /// + /// We probe with a trivial 1-qubit amplitude encode inside `catch_unwind` and + /// treat any panic or `Err` as "no functional GPU". A real GPU that errors + /// here is a genuine bug, which the `.expect(...)` calls below surface loudly. + fn cuda_is_functional() -> bool { + // `cudarc`'s dlopen failure panics with a long message; silence the + // default hook just for the probe so a skipped run isn't mistaken for a + // failure. Serialized because the panic hook is process-global. + static PROBE_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); + let _guard = PROBE_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + let prev_hook = std::panic::take_hook(); + std::panic::set_hook(Box::new(|_| {})); + let probe = std::panic::catch_unwind(|| { + let engine = common::qdp_engine_with_precision(Precision::Float32)?; + let dlpack = engine + .encode_batch_f32(&[1.0_f32, 0.0], 1, 2, 1, "amplitude") + .ok()?; + unsafe { common::take_deleter_and_delete(dlpack) }; + Some(()) + }); + std::panic::set_hook(prev_hook); + matches!(probe, Ok(Some(()))) + } + + /// Read the same data from an f32 and an f64 Parquet column, encode at the + /// matching precision, and return the minimum per-sample cross-precision + /// fidelity. `None` when no functional GPU is available. + fn parquet_amplitude_fidelity(num_qubits: usize) -> Option { + if !cuda_is_functional() { + return None; + } + let engine_f64 = common::qdp_engine_with_precision(Precision::Float64)?; + let engine_f32 = common::qdp_engine_with_precision(Precision::Float32)?; + let device = common::cuda_device()?; + + let state_dim = 1usize << num_qubits; + let num_samples = 4; + let sample_size = state_dim; + + let data_f64 = make_amplitude_data(num_samples, sample_size); + let (f32_file, f64_file) = write_pair(&data_f64, sample_size); + + // f32 column → f32 reader → f32 kernel + let mut reader_f32 = + ParquetReader::::new(&f32_file.0, None, NullHandling::FillZero).unwrap(); + let (host_in_f32, _, _) = reader_f32.read_batch().unwrap(); + let dlpack_f32 = engine_f32 + .encode_batch_f32( + &host_in_f32, + num_samples, + sample_size, + num_qubits, + "amplitude", + ) + .expect("F32 encode_batch should succeed"); + + // f64 column → f64 reader → f64 kernel + let mut reader_f64 = + ParquetReader::::new(&f64_file.0, None, NullHandling::FillZero).unwrap(); + let (host_in_f64, _, _) = reader_f64.read_batch().unwrap(); + let dlpack_f64 = engine_f64 + .encode_batch( + &host_in_f64, + num_samples, + sample_size, + num_qubits, + "amplitude", + ) + .expect("F64 encode_batch should succeed"); + + let f64_tensor = unsafe { &(*dlpack_f64).dl_tensor }; + let f32_tensor = unsafe { &(*dlpack_f32).dl_tensor }; + + let total_elements = num_samples * state_dim; + let host_f64 = + download_complex_f64(&device, f64_tensor.data as *const _, total_elements).unwrap(); + let host_f32 = + download_complex_f32(&device, f32_tensor.data as *const _, total_elements).unwrap(); + + let mut min_fidelity = 1.0_f64; + for s in 0..num_samples { + let off = s * state_dim * 2; + let sample_f64 = &host_f64[off..off + state_dim * 2]; + let sample_f32 = &host_f32[off..off + state_dim * 2]; + let f = fidelity_cross_precision(sample_f32, sample_f64).unwrap(); + if f < min_fidelity { + min_fidelity = f; + } + } + + unsafe { + common::take_deleter_and_delete(dlpack_f64); + common::take_deleter_and_delete(dlpack_f32); + } + + Some(min_fidelity) + } + + /// Run the Parquet f32-vs-f64 amplitude case at `num_qubits`, skipping when + /// no GPU is available. Threshold matches `gpu_fidelity.rs` (1e-3 at 8-16). + fn assert_amplitude_fidelity(num_qubits: usize) { + let Some(fidelity) = parquet_amplitude_fidelity(num_qubits) else { + println!("SKIP: No GPU available"); + return; + }; + println!("Parquet F32 vs F64 fidelity @ {num_qubits} qubits: {fidelity:.10}"); + assert!( + fidelity > 1.0 - 1e-3, + "Fidelity too low at {num_qubits} qubits: {fidelity}" + ); + } + + #[test] + fn test_parquet_f32_vs_f64_amplitude_8_qubits() { + assert_amplitude_fidelity(8); + } + + #[test] + fn test_parquet_f32_vs_f64_amplitude_12_qubits() { + assert_amplitude_fidelity(12); + } + + #[test] + fn test_parquet_f32_vs_f64_amplitude_16_qubits() { + assert_amplitude_fidelity(16); + } +} diff --git a/qdp/qdp-python/benchmark/README.md b/qdp/qdp-python/benchmark/README.md index d30715a1c8..6d11bfdf9b 100644 --- a/qdp/qdp-python/benchmark/README.md +++ b/qdp/qdp-python/benchmark/README.md @@ -9,6 +9,8 @@ scripts: that measures vectors/sec across Mahout, PennyLane, and Qiskit. - `benchmark_latency.py`: Data-to-State latency benchmark (CPU RAM -> GPU VRAM). - `benchmark_phase.py`: GPU phase encoding latency benchmark (batch encode timing). +- `benchmark_parquet_f32.py`: end-to-end f32 vs f64 Parquet pipeline throughput + (vectors/sec and f32/f64 speedup); see [Parquet f32 vs f64](#parquet-f32-vs-f64-throughput). ## Quick Start @@ -34,10 +36,32 @@ uv run --project qdp/qdp-python python qdp/qdp-python/benchmark/benchmark_e2e.py uv run --project qdp/qdp-python python qdp/qdp-python/benchmark/benchmark_latency.py uv run --project qdp/qdp-python python qdp/qdp-python/benchmark/benchmark_throughput.py uv run --project qdp/qdp-python python qdp/qdp-python/benchmark/benchmark_phase.py +uv run --project qdp/qdp-python python qdp/qdp-python/benchmark/benchmark_parquet_f32.py ``` This keeps all benchmark dependencies in the unified repo root venv (`mahout/.venv`). +## Parquet f32 vs f64 throughput + +`benchmark_parquet_f32.py` reads the same amplitude data from a Parquet file +through `QuantumDataLoader` at `dtype("float32")` and `dtype("float64")`, and +reports vectors/sec for each plus the f32/f64 speedup (issue #1342). The +expected win is ~25-35% when the Parquet column is native f32. + +```bash +uv run --project qdp/qdp-python python qdp/qdp-python/benchmark/benchmark_parquet_f32.py \ + --qubits 12 --batches 200 --batch-size 64 +``` + +By default two temporary files holding the same data — a native +`FixedSizeList` and a native `FixedSizeList` — are generated +and each is read at its matching dtype, so the f32 run measures the zero-copy +native-f32 path (not a f64→f32 cast). Pass `--parquet PATH` to benchmark your +own file at both dtypes instead. The generated data is materialized in RAM and +on disk and grows as `batches * batch_size * 2^qubits`, so raise `--qubits` with +that cost in mind. Requires an NVIDIA GPU (the native file loader is Linux + +CUDA only); not run in CI. + ## Manual Setup If you prefer to set up manually (or if `make benchmark` is not available): diff --git a/qdp/qdp-python/benchmark/benchmark_parquet_f32.py b/qdp/qdp-python/benchmark/benchmark_parquet_f32.py new file mode 100644 index 0000000000..396836b423 --- /dev/null +++ b/qdp/qdp-python/benchmark/benchmark_parquet_f32.py @@ -0,0 +1,215 @@ +#!/usr/bin/env python3 +# +# 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. + +"""End-to-end f32 vs f64 Parquet pipeline throughput benchmark (issue #1342). + +Reads amplitude data from a Parquet file through ``QuantumDataLoader`` at +``dtype("float32")`` and ``dtype("float64")`` and reports encoded +vectors/second for each, plus the f32/f64 speedup. The expected win is +~25-35% when the Parquet column is native f32 (see #1338). + +To actually exercise the native f32 path, the default (no ``--parquet``) run +generates *two* files holding the same logical data — one native +``FixedSizeList`` and one native ``FixedSizeList`` — and reads +each at its matching dtype, so the f32 run is the zero-copy native-f32 read, not +a f64→f32 cast. With ``--parquet PATH`` the same user file is read at both +dtypes (a native-f32 column hits the zero-copy path; a f64 column is cast to f32 +on read). + +This is a GPU/Linux-only benchmark: ``QuantumDataLoader``'s native file loader is +only available on Linux with CUDA, mirroring ``benchmark_phase.py``. It is NOT +wired into CI; run it locally on a GPU box. + +Run from the repo root:: + + uv run --project qdp/qdp-python \\ + python qdp/qdp-python/benchmark/benchmark_parquet_f32.py \\ + --qubits 12 --batches 200 --batch-size 64 + +The generated data is materialized in RAM and on disk and grows as +``batches * batch_size * 2^qubits``, so raise ``--qubits`` with that cost in mind. +""" + +from __future__ import annotations + +import argparse +import os +import tempfile +import time + +import numpy as np +import pyarrow as pa +import pyarrow.parquet as pq +from qumat_qdp import QuantumDataLoader + + +def _generate_parquet( + path: str, + total_rows: int, + sample_size: int, + np_dtype: type[np.floating], + pa_type: pa.DataType, +) -> None: + """Write a ``FixedSizeList`` Parquet file of random data. + + Each call generates its own data from a fixed seed, so the f32 and f64 files + hold the same logical values without the caller keeping both large host + arrays alive at once. A FixedSizeList has no offsets buffer, so this also + sidesteps the int32 offset overflow a variable ``List`` hits once + ``total_rows * sample_size`` exceeds 2^31 (large --qubits). + """ + flat = np.random.default_rng(0).random(total_rows * sample_size) + if np_dtype != np.float64: + flat = flat.astype(np_dtype) + values = pa.array(flat, type=pa_type) + list_array = pa.FixedSizeListArray.from_arrays(values, sample_size) + pq.write_table(pa.table({"data": list_array}), path) + + +def run_loader( + path: str, + dtype: str, + num_qubits: int, + total_batches: int, + batch_size: int, + encoding_method: str, +) -> tuple[float, float]: + """Iterate the loader at the given dtype; return (duration_sec, vec/s). + + The native file loader reads the *entire* file regardless of ``--batches``: + ``total_batches`` is ignored for file sources (only ``--batch-size`` sets + batch granularity), and iteration runs until EOF. So the loop processes + exactly the file's Parquet row count, and ``num_rows / elapsed`` is the true + throughput for any ``--parquet`` file -- not just the generated case where + ``num_rows == batches * batch_size``. (Do not clamp to ``batches * + batch_size``: for a file larger than that, the whole file is still read, so + clamping would divide the full read time by an undercount.) + """ + total_vectors = pq.read_metadata(path).num_rows + loader = ( + QuantumDataLoader(device_id=0) + .qubits(num_qubits) + .encoding(encoding_method) + .batches(total_batches, size=batch_size) + .dtype(dtype) + .source_file(path) + ) + start = time.perf_counter() + for _ in loader: # drive the pipeline; the encoded batch is not needed + pass + elapsed = max(time.perf_counter() - start, 1e-9) + return elapsed, total_vectors / elapsed + + +def main() -> None: + parser = argparse.ArgumentParser( + description="f32 vs f64 Parquet pipeline throughput benchmark" + ) + # Default sizes are deliberately modest: the generated data is materialized + # in host RAM and written to disk (one f32 + one f64 file), so it grows as + # batches * batch_size * 2^qubits. 12 qubits keeps the default run ~0.4 GB; + # raise --qubits with that cost in mind. + parser.add_argument("--qubits", type=int, default=12) + parser.add_argument( + "--batches", + type=int, + default=200, + help="Number of batches. Sizes the generated data (batches * batch_size " + "rows); ignored with --parquet, where the whole file is always read.", + ) + parser.add_argument("--batch-size", type=int, default=64) + parser.add_argument("--encoding", type=str, default="amplitude") + parser.add_argument("--trials", type=int, default=3) + parser.add_argument( + "--parquet", + type=str, + default=None, + help="Existing Parquet file to benchmark at both dtypes; " + "if omitted, native f32 and f64 temp files are generated.", + ) + args = parser.parse_args() + + sample_size = 1 << args.qubits + total_rows = args.batches * args.batch_size + + tmp_paths: list[str] = [] + # Map each dtype to the file it should read. For the generated case each + # dtype reads its own native-precision file (no cast); for --parquet both + # read the user's single file. + paths: dict[str, str] + if args.parquet is None: + f64_fd, f64_path = tempfile.mkstemp(suffix="_bench_f64.parquet") + f32_fd, f32_path = tempfile.mkstemp(suffix="_bench_f32.parquet") + os.close(f64_fd) + os.close(f32_fd) + # Register for cleanup before generating, so a failure mid-generation + # (e.g. MemoryError on large --qubits, or disk-full) still removes them. + tmp_paths = [f64_path, f32_path] + paths = {"float64": f64_path, "float32": f32_path} + else: + paths = {"float64": args.parquet, "float32": args.parquet} + + try: + if args.parquet is None: + print( + f"Generating {total_rows} rows x {sample_size} cols (native f32 + f64)" + ) + _generate_parquet( + paths["float64"], total_rows, sample_size, np.float64, pa.float64() + ) + _generate_parquet( + paths["float32"], total_rows, sample_size, np.float32, pa.float32() + ) + + print("=" * 70) + print("Parquet f32 vs f64 pipeline throughput") + print( + f" qubits={args.qubits}, batches={args.batches}, " + f"batch_size={args.batch_size}, encoding={args.encoding}" + ) + print("=" * 70) + + results: dict[str, float] = {} + for dtype in ("float64", "float32"): + vps_trials: list[float] = [] + for t in range(args.trials): + dur, vps = run_loader( + paths[dtype], + dtype, + args.qubits, + args.batches, + args.batch_size, + args.encoding, + ) + vps_trials.append(vps) + print(f" [{dtype}] trial {t + 1}: {dur:.4f} s, {vps:.1f} vec/s") + results[dtype] = sorted(vps_trials)[len(vps_trials) // 2] + print(f" [{dtype}] median: {results[dtype]:.1f} vec/s") + + print("-" * 70) + if results["float64"] <= 0.0: + print(" No vectors encoded (empty file?); skipping speedup.") + else: + speedup = results["float32"] / results["float64"] + print(f" f32/f64 speedup: {speedup:.3f}x ({(speedup - 1.0) * 100:+.1f}%)") + finally: + for p in tmp_paths: + os.remove(p) + + +if __name__ == "__main__": + main()