Skip to content

Speed up multihost exchange #5819

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
blp merged 8 commits into from
Mar 25, 2026
Merged

Speed up multihost exchange #5819

Changes from 1 commit
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
c270207
[nexmark] Remove unused imports.
blp Mar 19, 2026
d0e131c
[dbsp] Tiny code cleanup in try_send_all() callers.
blp Mar 12, 2026
33d32e0
[adapters] Simplify async_samply_profile with tokio::fs::read.
blp Mar 18, 2026
a1baa04
[adapters] Make the host layout available from the Controller.
blp Mar 24, 2026
63a1aa0
Improve support for custom marker ranges in CPU profiles.
blp Mar 9, 2026
3fdae5a
[dbsp] Use new protocol for exchanging data among hosts.
blp Mar 20, 2026
b5df7a4
[adapters] Indicate host ordinal in multihost profiles.
blp Mar 20, 2026
7f2a613
[adapters] Fix adhoc scan implementation for multihost.
blp Mar 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
[adapters] Fix adhoc scan implementation for multihost. 
When scanning a table produces multiple batches, it's important to feed
all of the batches into a single `StreamWriter`.  Until now, the code here
used a separate `StreamWriter` for each batch, but that didn't work
properly.

This wasn't noticed until now because the tests only produced small
outputs that had only one batch.

Signed-off-by: Ben Pfaff <blp@feldera.com>
  • Loading branch information
@blp
blp committed Mar 24, 2026
commit 7f2a613bbe4a686a8bca7fbe8a593188b4b95b33
25 changes: 13 additions & 12 deletions crates/adapters/src/server.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2579,20 +2579,21 @@ async fn coordination_adhoc_scan(
// likely.
let first_batch = match stream.next().await {
Some(Err(error)) => return Err(error.into()),
other => other,
}
.into_iter();
other => other.into_iter(),
};

let schema = stream.schema();
Ok(HttpResponseBuilder::new(StatusCode::OK).streaming(
futures_util::stream::iter(first_batch).chain(stream).map(
move |batch| -> Result<Bytes, PipelineError> {
let mut writer = StreamWriter::try_new(Vec::new(), &schema)?;
writer.write(&batch?)?;
Ok(writer.into_inner()?.into())
},
),
))
let response_stream = async_stream::try_stream! {
let mut writer = StreamWriter::try_new(Vec::new(), &schema)?;
let mut stream = futures_util::stream::iter(first_batch).chain(stream);
while let Some(batch) = stream.next().await {
writer.write(&batch?)?;
yield Bytes::copy_from_slice(writer.get_ref().as_slice());
writer.get_mut().clear();
}
yield writer.into_inner()?.into();
};
Ok(HttpResponseBuilder::new(StatusCode::OK).streaming::<_, PipelineError>(response_stream))
}

/// Stream the set of incomplete labels.
Expand Down
Loading