rich7420 commented on code in PR #1407:
URL: https://github.com/apache/mahout/pull/1407#discussion_r3448690818
##########
qdp/qdp-core/src/pipeline_runner.rs:
##########
@@ -1365,4 +1474,190 @@ mod tests {
"normalize() must set prefetch_depth > 0"
);
}
+
+ //
-------------------------------------------------------------------------
+ // dtype file-load tests
+ //
+ // These tests verify that PipelineConfig.dtype is respected when loading
+ // from file sources. They stop at the BatchData variant boundary — the
+ // encode kernel (encode_batch_f32_for_pipeline) is CUDA-gated and cannot
+ // be exercised in CPU-only CI. BatchData::F32 is the observable proxy
that
+ // confirms the f32 kernel would be called on a GPU host; this mirrors the
+ // existing convention in test_synthetic_producer_f32_*.
+ //
-------------------------------------------------------------------------
+
+ mod dtype_file_tests {
+ use super::*;
+ use arrow::array::{ArrayRef, FixedSizeListBuilder, Float32Builder,
RecordBatch};
+ use arrow::datatypes::{DataType, Field, Schema};
+ use parquet::arrow::ArrowWriter;
+ use std::fs;
+ use std::sync::Arc;
+
+ fn write_f32_parquet(path: &std::path::Path) {
+ // 8 samples, each 4 features — matches amplitude encoding with 2
qubits (2^2=4)
+ let item_field = Arc::new(Field::new("item", DataType::Float32,
true));
+ let list_field = Field::new("data",
DataType::FixedSizeList(item_field, 4), true);
+ let schema = Arc::new(Schema::new(vec![list_field]));
+ let mut builder = FixedSizeListBuilder::new(Float32Builder::new(),
4);
+ for _ in 0..8 {
+ builder.values().append_slice(&[0.25_f32, 0.5, 0.75, 1.0]);
+ builder.append(true);
+ }
+ let array = Arc::new(builder.finish()) as ArrayRef;
+ let batch = RecordBatch::try_new(schema.clone(),
vec![array]).unwrap();
+ let file = fs::File::create(path).unwrap();
+ let mut writer = ArrowWriter::try_new(file, schema, None).unwrap();
+ writer.write(&batch).unwrap();
+ writer.close().unwrap();
+ }
+
+ static FILE_COUNTER: std::sync::atomic::AtomicUsize =
+ std::sync::atomic::AtomicUsize::new(0);
+
+ fn temp_parquet_path(tag: &str) -> std::path::PathBuf {
+ let n = FILE_COUNTER.fetch_add(1,
std::sync::atomic::Ordering::Relaxed);
+ std::env::temp_dir().join(format!(
+ "mahout_pipeline_dtype_{tag}_{pid}_{n}.parquet",
+ pid = std::process::id(),
+ ))
+ }
+
+ #[test]
+ fn test_read_file_by_extension_f32_parquet_returns_f32_batch_data() {
+ let path = temp_parquet_path("f32");
+ write_f32_parquet(&path);
+ let result = read_file_by_extension(&path, NullHandling::FillZero,
Precision::Float32);
+ let _ = fs::remove_file(&path);
+ let (batch_data, num_samples, sample_size) = result.unwrap();
+ assert!(
+ matches!(batch_data, BatchData::F32(_)),
Review Comment:
nit (non-blocking): these new tests only check the `BatchData` *variant*,
not the values — so a batch that comes back as `F32` but with zeroed/garbled
contents would still pass green. Might be worth reading one element back (e.g.
`assert` the first value ≈ `0.25`) so it's a real check, like
`test_synthetic_producer_f32_amplitude` does. Same gap means the
`cast_f64_to_batch_data` path (arrow/npy/pt/pb → f32) and the f64-parquet→f32
cross-cast don't have coverage yet. 👍
##########
qdp/qdp-core/src/pipeline_runner.rs:
##########
@@ -492,26 +628,42 @@ impl PipelineIterator {
config.encoding.as_str()
)));
}
- if data.len() != num_samples * sample_size {
+ if batch_data.len() != num_samples * sample_size {
return Err(MahoutError::InvalidInput(format!(
"File data length {} is not num_samples ({}) * sample_size
({})",
- data.len(),
+ batch_data.len(),
num_samples,
sample_size
)));
}
- let producer = InMemoryProducer {
- data,
- cursor: 0,
- sample_size,
- batch_size: config.batch_size,
- num_qubits: config.num_qubits as usize,
- batches_yielded: 0,
- batch_limit,
- };
let prefetch_depth = config.prefetch_depth;
- let (rx, recycle_tx, _producer_handle) = spawn_producer(producer,
prefetch_depth)?;
+ let (rx, recycle_tx, _producer_handle) = match batch_data {
+ BatchData::F32(data) => spawn_producer(
Review Comment:
nit (totally optional): these two arms are byte-for-byte identical apart
from the `<f32>`/`<f64>` type param, and the streaming path already has
`build_streaming_producer<T>` — a little `build_inmemory_producer<T>` helper
here would kill the copy-paste and mirror that pattern. Just caught my eye
while reading.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
