You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mahout.apache.org by "andrewmusselman (via GitHub)" <gi...@apache.org> on 2026/10/20 17:58:10 UTC
[PR] [QDP] Unify PipelineIterator constructors with a Source enum (#1434) [mahout]
andrewmusselman opened a new pull request, #1557:
URL: https://github.com/apache/mahout/pull/1557
### Purpose of PR
`PipelineIterator` had three constructors that each repeated normalize, producer spawn and struct setup. This adds a `Source` enum and one `PipelineIterator::new(engine, source, config)`, moves the PyO3 loaders to it, and keeps the old constructors as deprecated wrappers so out-of-tree Rust callers keep compiling. Producer logic is unchanged.
### Related Issues or PRs
Closes #1434
First of three: the #1435 PR builds on this one.
### Changes Made
- `qdp-core/src/pipeline_runner.rs`: `Source::{Synthetic { total_batches, seed }, File { .., batch_limit }, FileStreaming { .., batch_limit }}` and `PipelineIterator::new`.
- `batch_limit` lives in the file variants, since synthetic runs would ignore it in `PipelineConfig`.
- `Source::Synthetic` is authoritative: `new` copies `total_batches`/`seed` into the config before `normalize()`.
- Construction logic in a private `spawn_source` so it is testable on CPU-only machines; old constructor bodies moved verbatim into `spawn_file_producer` / `spawn_streaming_file_producer`.
- `Source` re-exported from `qdp-core/src/lib.rs`.
- `qdp-python/src/engine.rs`: `create_synthetic_loader`, `create_file_loader`, `create_streaming_file_loader` use `new`; Python signatures unchanged (`batch_limit=None` still means `usize::MAX`).
- `new_synthetic`, `new_from_file`, `new_from_file_streaming` become `#[deprecated]` one-line wrappers.
- Tests: one per `Source` variant through `spawn_source`, streaming non-`.parquet` rejection, and a GPU test of `new` and the deprecated wrappers for every variant (skips without a GPU).
### Testing
- `cargo test -p qdp-core --lib`: 102 passed, 3 failed (main 97/3; same 3 GPU-only tests; +5 new).
- `--test parquet_f32 --test reader`: 7 + 3 passed, unchanged.
- `cargo clippy -p qdp-core -p qdp-python --all-targets -- -D warnings` clean at each of the 3 commits; `cargo check -p qdp-python` (also `--features remote-io`) and `cargo fmt --check` clean.
- Not verified: the GPU test on a real GPU (it skipped), the Python test suite (wheel not built), `--all-features` clippy (needs libtorch).
### Merging
Conflicts with the stand-alone #1436 VecDeque patch in `pipeline_runner.rs`; an "after #1435" version of that patch exists.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
https://claude.ai/code/session_01GuXmXRXdNF43h2FoQcNAZb
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@mahout.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [QDP] Unify PipelineIterator constructors with a Source enum (#1434) [mahout]
Posted by "400Ping (via GitHub)" <gi...@apache.org>.
400Ping commented on code in PR #1557:
URL: https://github.com/apache/mahout/pull/1557#discussion_r4423125872
##########
qdp/qdp-core/src/pipeline_runner.rs:
##########
@@ -645,123 +645,206 @@ where
spawn_producer(producer, config.prefetch_depth)
}
+/// Where a [`PipelineIterator`] gets its samples from.
+///
+/// Every knob that only makes sense for one kind of source lives in that variant, so
+/// [`PipelineConfig`] stays the set of settings shared by all sources (qubits, batch size,
+/// encoding, dtype, null handling, prefetch depth).
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum Source {
Review Comment:
Since this is re-exported from `qdp_core` and will likely gain variants (remote sources, in-memory tensors), `#[non_exhaustive]` now avoids a semver break later.
##########
qdp/qdp-core/src/pipeline_runner.rs:
##########
@@ -1910,4 +1993,195 @@ mod tests {
);
}
}
+
+ // -------------------------------------------------------------------------
+ // Source enum (#1434)
+ //
+ // `PipelineIterator::new` = `spawn_source` + storing a `QdpEngine`. Creating an
+ // engine needs CUDA, so the per-variant tests drive `spawn_source` directly and read
+ // the batches off the prefetch channel; the GPU-only test at the end covers `new`
+ // itself and skips when no functional GPU is present.
+ // -------------------------------------------------------------------------
+
+ mod source_tests {
+ use super::dtype_file_tests::{temp_parquet_path, write_f32_parquet};
+ use super::*;
+ use std::fs;
+
+ /// Drain every batch the producer thread sends, then join it.
+ fn collect(handles: ProducerHandles) -> Vec<PrefetchedBatch> {
+ let (rx, _recycle_tx, handle) = handles;
+ let batches: Vec<_> = rx.iter().map(|b| b.unwrap()).collect();
+ handle.join().unwrap();
+ batches
+ }
+
+ #[test]
+ fn test_source_synthetic_overrides_config_batches_and_seed() {
+ // Source::Synthetic is authoritative over config.total_batches / config.seed.
+ let mut config = PipelineConfig {
+ total_batches: 999,
+ seed: None,
+ num_qubits: 3,
+ batch_size: 4,
+ encoding: Encoding::Amplitude,
+ ..Default::default()
+ };
+ let source = Source::Synthetic {
+ total_batches: 3,
+ seed: Some(7),
+ };
+ let batches = collect(spawn_source(source, &mut config).unwrap());
+
+ assert_eq!(batches.len(), 3);
+ assert_eq!((config.total_batches, config.seed), (3, Some(7)));
+ assert!(config.prefetch_depth > 0, "spawn_source must normalize()");
+ let expected_cfg = PipelineConfig {
+ seed: Some(7),
+ ..config.clone()
+ };
+ for (i, b) in batches.iter().enumerate() {
+ assert_eq!(b.batch_n, 4);
+ assert_eq!(b.data, BatchData::F64(generate_batch(&expected_cfg, i, 8)));
+ }
+ }
+
+ #[test]
+ fn test_source_file_honours_batch_limit_and_dtype() {
+ // 8 samples x 4 features; batch_size 3 -> 3,3,2 until EOF.
+ let path = temp_parquet_path("source_file");
+ write_f32_parquet(&path);
+ let base = PipelineConfig {
+ num_qubits: 2,
+ batch_size: 3,
+ encoding: Encoding::Amplitude,
+ dtype: Precision::Float32,
+ ..Default::default()
+ };
+ let run = |batch_limit| {
+ let source = Source::File {
+ path: path.clone(),
+ batch_limit,
+ };
+ spawn_source(source, &mut base.clone()).map(collect)
+ };
+ let all = run(usize::MAX);
+ let limited = run(2);
+ let _ = fs::remove_file(&path);
+
+ let all = all.unwrap();
+ let sizes: Vec<_> = all.iter().map(|b| b.batch_n).collect();
+ assert_eq!(sizes, vec![3, 3, 2]);
+ assert!(matches!(all[0].data, BatchData::F32(_)));
+ assert_eq!(limited.unwrap().len(), 2, "batch_limit must cap batches");
+ }
+
+ #[test]
+ fn test_source_file_streaming_honours_batch_limit_and_dtype() {
+ let path = temp_parquet_path("source_stream");
+ write_f32_parquet(&path);
+ let base = PipelineConfig {
+ num_qubits: 2,
+ batch_size: 3,
+ encoding: Encoding::Amplitude,
+ dtype: Precision::Float32,
+ ..Default::default()
+ };
+ let run = |batch_limit| {
+ let source = Source::FileStreaming {
+ path: path.clone(),
+ batch_limit,
+ };
+ spawn_source(source, &mut base.clone()).map(collect)
+ };
+ let all = run(usize::MAX);
+ let limited = run(1);
+ let _ = fs::remove_file(&path);
+
+ let all = all.unwrap();
+ let sizes: Vec<_> = all.iter().map(|b| b.batch_n).collect();
+ assert_eq!(sizes, vec![3, 3, 2]);
+ match &all[0].data {
+ BatchData::F32(v) => assert_eq!(&v[..4], &[0.25_f32, 0.5, 0.75, 1.0]),
+ other => panic!("expected F32, got {other:?}"),
+ }
+ assert_eq!(limited.unwrap().len(), 1, "batch_limit must cap batches");
+ }
+
+ #[test]
+ fn test_source_file_streaming_rejects_non_parquet() {
+ let source = Source::FileStreaming {
+ path: PathBuf::from("data.npy"),
+ batch_limit: usize::MAX,
+ };
+ let err = spawn_source(source, &mut PipelineConfig::default()).err();
+ assert!(matches!(err, Some(MahoutError::InvalidInput(_))));
+ }
+
+ /// A `QdpEngine`, or `None` when CUDA is unusable. Without a driver `cudarc`
+ /// panics instead of returning `Err`, so probe inside `catch_unwind`.
+ fn try_engine() -> Option<QdpEngine> {
+ static LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
+ let _guard = LOCK.lock().unwrap_or_else(|e| e.into_inner());
+ let prev = std::panic::take_hook();
+ std::panic::set_hook(Box::new(|_| {}));
Review Comment:
Swapping the global panic hook affects every test running in parallel in this window (their panic messages are swallowed). `catch_unwind` alone, as `tests/parquet_f32_fidelity.rs:143` already does for its GPU probe, would avoid that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@mahout.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org