feat(mohu-io): implement Arrow IPC file and stream read/write - #264
feat(mohu-io): implement Arrow IPC file and stream read/write#264madhu-mitha-e wants to merge 4 commits into
Conversation
Closes mohu-org#35 Signed-off-by: madhu-mitha-e <e.madhumitha06052006@gmail.com>
|
Warning Review limit reached
More reviews will be available in 36 minutes and 57 seconds. Learn how PR review limits work. Your organization has run out of usage credits. Purchase more in the billing tab. ⌛ How to resolve this issue?After more reviews become available, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans include higher PR review limits than trial, open-source, and free plans. In all cases, reviews become available again over time. During sustained high-volume PR review activity, CodeRabbit may temporarily slow when the next review becomes available. Please see our Fair Usage Limits Policy for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (10)
📝 WalkthroughWalkthroughThis PR adds Arrow IPC (Inter-Process Communication) support to mohu by introducing serialization and deserialization functions for both file and streaming formats. The implementation enables bidirectional conversions between mohu Buffers and Arrow arrays, with proper error handling and comprehensive unit tests covering multiple data types. ChangesArrow IPC file and stream support
Estimated code review effort🎯 2 (Simple) | ⏱️ ~10 minutes Possibly related issues
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
PR Check Summary
CI will run: build, test, clippy, fmt, cargo-deny, DCO, semver. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@crates/mohu-io/src/arrow.rs`:
- Around line 318-329: Tests compare floating-point slices using assert_eq!
which is brittle; replace those exact-equality assertions with mohutesting's
approximate checker by importing mohutesting::approx::assert_allclose and
changing the assert_eq! calls in the float tests (e.g.,
round_trip_ipc_stream_f32, round_trip_ipc_stream_f64 and the other float slice
assertion around line 353 that checks result[i].as_slice::<f32|f64>()) to use
assert_allclose(result[..].as_slice::<f32|f64>().unwrap(), &[...] ) so the tests
use approximate float comparison when validating
read_ipc_stream/write_ipc_stream and Buffer round-trip values.
- Around line 181-183: Wrap the Arrow operation failures with a context
breadcrumb before converting to MohuError: for each site (e.g.,
RecordBatch::try_new -> currently mapped to MohuError::ArrowSchema, and the
various ArrowIpc reads/writes mapped to MohuError::ArrowIpc) insert a
.with_context(|| "attempting to <brief op description>") or .context("attempting
to <brief op description>") call on the Result (for example, before the final
.map_err(|e| MohuError::ArrowSchema(e.to_string())) for RecordBatch::try_new add
.with_context(|| "creating RecordBatch via RecordBatch::try_new") so the error
carries what was being attempted, and do the same for each ArrowIpc site
(reads/writes) describing the specific IPC operation being attempted before
mapping to MohuError::ArrowIpc.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 3d0eac1c-f866-470d-908c-65c32106f479
📒 Files selected for processing (2)
crates/mohu-io/Cargo.tomlcrates/mohu-io/src/arrow.rs
| RecordBatch::try_new(schema, arrays) | ||
| .map_err(|e| MohuError::ArrowSchema(e.to_string())) | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Verify all direct Arrow IPC/schema map_err conversions lacking context in this file.
rg -nP 'map_err\(\|e\|\s*MohuError::Arrow(Ipc|Schema)\(e\.to_string\(\)\)\)' crates/mohu-io/src/arrow.rsRepository: mohu-org/mohu
Length of output: 785
Add operation .context(...)/.with_context(...) to Arrow error conversions in crates/mohu-io/src/arrow.rs.
The file contains multiple direct map_err(|e| MohuError::ArrowSchema/ArrowIpc(e.to_string())) conversions without an operation breadcrumb (e.g., RecordBatch::try_new at line 182, and ArrowIpc sites at lines 208, 212, 231, 235, 259, 263, 267, 290, 294, 298). Wrap these failures with .context("what was being attempted") (or .with_context(|| ...)) so errors include the attempted Arrow IPC/schema operation.
Example (current state)
RecordBatch::try_new(schema, arrays)
.map_err(|e| MohuError::ArrowSchema(e.to_string()))🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@crates/mohu-io/src/arrow.rs` around lines 181 - 183, Wrap the Arrow operation
failures with a context breadcrumb before converting to MohuError: for each site
(e.g., RecordBatch::try_new -> currently mapped to MohuError::ArrowSchema, and
the various ArrowIpc reads/writes mapped to MohuError::ArrowIpc) insert a
.with_context(|| "attempting to <brief op description>") or .context("attempting
to <brief op description>") call on the Result (for example, before the final
.map_err(|e| MohuError::ArrowSchema(e.to_string())) for RecordBatch::try_new add
.with_context(|| "creating RecordBatch via RecordBatch::try_new") so the error
carries what was being attempted, and do the same for each ArrowIpc site
(reads/writes) describing the specific IPC operation being attempted before
mapping to MohuError::ArrowIpc.
| assert_eq!(result[0].as_slice::<f32>().unwrap(), &[1.0_f32, 2.0, 3.0, 4.0]); | ||
| } | ||
|
|
||
| #[test] | ||
| fn round_trip_ipc_stream_f64() { | ||
| let buf = Buffer::from_slice(&[10.0_f64, 20.0, 30.0]).unwrap(); | ||
| let mut bytes = Vec::new(); | ||
| write_ipc_stream(&mut bytes, &[buf.clone()]).unwrap(); | ||
|
|
||
| let result = read_ipc_stream(Cursor::new(bytes)).unwrap(); | ||
| assert_eq!(result.len(), 1); | ||
| assert_eq!(result[0].as_slice::<f64>().unwrap(), &[10.0_f64, 20.0, 30.0]); |
There was a problem hiding this comment.
Replace float assert_eq! with assert_allclose.
Line 318, Line 329, and Line 353 compare float slices using exact equality. Please switch these to mohu_testing::approx::assert_allclose to satisfy test policy and avoid brittle comparisons.
Suggested test update
#[cfg(test)]
mod tests {
use super::*;
+ use mohu_testing::approx::assert_allclose;
use std::io::Cursor;
@@
- assert_eq!(result[0].as_slice::<f32>().unwrap(), &[1.0_f32, 2.0, 3.0, 4.0]);
+ assert_allclose(result[0].as_slice::<f32>().unwrap(), &[1.0_f32, 2.0, 3.0, 4.0], 1e-6, 1e-6);
@@
- assert_eq!(result[0].as_slice::<f64>().unwrap(), &[10.0_f64, 20.0, 30.0]);
+ assert_allclose(result[0].as_slice::<f64>().unwrap(), &[10.0_f64, 20.0, 30.0], 1e-12, 1e-12);
@@
- assert_eq!(result[1].as_slice::<f64>().unwrap(), &[4.0_f64, 5.0, 6.0]);
+ assert_allclose(result[1].as_slice::<f64>().unwrap(), &[4.0_f64, 5.0, 6.0], 1e-12, 1e-12);As per coding guidelines, "Float comparison must use mohu-testing::approx::assert_allclose — never == on floats."
Also applies to: 353-353
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@crates/mohu-io/src/arrow.rs` around lines 318 - 329, Tests compare
floating-point slices using assert_eq! which is brittle; replace those
exact-equality assertions with mohutesting's approximate checker by importing
mohutesting::approx::assert_allclose and changing the assert_eq! calls in the
float tests (e.g., round_trip_ipc_stream_f32, round_trip_ipc_stream_f64 and the
other float slice assertion around line 353 that checks
result[i].as_slice::<f32|f64>()) to use
assert_allclose(result[..].as_slice::<f32|f64>().unwrap(), &[...] ) so the tests
use approximate float comparison when validating
read_ipc_stream/write_ipc_stream and Buffer round-trip values.
Signed-off-by: madhu-mitha-e <e.madhumitha06052006@gmail.com>
39a386d to
92fbf3e
Compare
1 participant
Closes #35
What
Implement Arrow IPC file and streaming format support in
crates/mohu-io/src/arrow.rs:read_ipc_file— reads Arrow IPC file format (random-access, seekable)read_ipc_stream— reads Arrow IPC streaming format (sequential, no footer)write_ipc_file— writes buffers to Arrow IPC file formatwrite_ipc_stream— writes buffers to Arrow IPC streaming formatWhy
Without streaming IPC, mohu cannot participate in zero-copy Arrow pipelines where producers and consumers live in different processes or machines (Polars pipe, DuckDB query results, Arrow Flight protocol).
arrow = "54"was already in workspace deps.How
arrow::ipc::reader::{FileReader, StreamReader}andarrow::ipc::writer::{FileWriter, StreamWriter}Bufferbecomes one column in aRecordBatchfeatures = ["ipc"]tomohu-io/Cargo.tomlChecklist
cargo test --workspacepassescargo clippy --workspace -- -D warningspassescargo fmt --allappliedCHANGELOG.mdupdatedSummary by CodeRabbit
New Features