This is an automated email from the ASF dual-hosted git repository.
rich7420 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/mahout.git
The following commit(s) were added to refs/heads/main by this push:
new 58e4aff32 [Testing][QDP] f32 Parquet fidelity tests and benchmark
(#1422)
58e4aff32 is described below
commit 58e4aff32c52e115b8fffd66eb84756b74e303f3
Author: ChenChen Lai <[email protected]>
AuthorDate: Sat Jul 4 16:13:01 2026 +0800
[Testing][QDP] f32 Parquet fidelity tests and benchmark (#1422)
* f32 Parquet fidelity tests and benchmark
* fix error
* address comment
---
.github/workflows/python-testing.yml | 18 +-
qdp/qdp-core/tests/common/mod.rs | 77 ++++++-
qdp/qdp-core/tests/parquet_f32_fidelity.rs | 257 ++++++++++++++++++++++
qdp/qdp-python/benchmark/README.md | 24 ++
qdp/qdp-python/benchmark/benchmark_parquet_f32.py | 215 ++++++++++++++++++
5 files changed, 587 insertions(+), 4 deletions(-)
diff --git a/.github/workflows/python-testing.yml
b/.github/workflows/python-testing.yml
index 23061d612..9f025647a 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 7114ad613..241012d0f 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<Float32>` 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<Float64>` 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 000000000..58f40b9bd
--- /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<Float32>` and a `List<Float64>`
+//! column, read back through `ParquetReader::<f32>` / `ParquetReader::<f64>`,
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<f64> {
+ 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<Float32>` and `List<Float64>` 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<f32> = 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::<f32>::new(&f32_file.0, None,
NullHandling::FillZero).unwrap();
+ let (data32, n32, s32) = reader_f32.read_batch().unwrap();
+
+ let mut reader_f64 =
+ ParquetReader::<f64>::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<f64> {
+ 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::<f32>::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::<f64>::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 d30715a1c..6d11bfdf9 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<Float32>` and a native `FixedSizeList<Float64>` — 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 000000000..396836b42
--- /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<Float32>`` and one native ``FixedSizeList<Float64>`` — 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<pa_type, sample_size>`` 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()
