Skip to content
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

Make execute_stream functions sync #4608

Merged
merged 1 commit into from
Dec 14, 2022
Merged

Make execute_stream functions sync #4608

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
4 changes: 2 additions & 2 deletions datafusion/core/src/dataframe.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -457,7 +457,7 @@ impl DataFrame {
pub async fn execute_stream(&self) -> Result<SendableRecordBatchStream> {
let plan = self.create_physical_plan().await?;
let task_ctx = Arc::new(TaskContext::from(&self.session_state.read().clone()));
execute_stream(plan, task_ctx).await
execute_stream(plan, task_ctx)
}

/// Executes this DataFrame and collects all results into a vector of vector of RecordBatch
Expand Down Expand Up @@ -498,7 +498,7 @@ impl DataFrame {
) -> Result<Vec<SendableRecordBatchStream>> {
let plan = self.create_physical_plan().await?;
let task_ctx = Arc::new(TaskContext::from(&self.session_state.read().clone()));
execute_stream_partitioned(plan, task_ctx).await
execute_stream_partitioned(plan, task_ctx)
}

/// Returns the schema describing the output of this DataFrame in terms of columns returned,
Expand Down
8 changes: 4 additions & 4 deletions datafusion/core/src/physical_plan/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -406,12 +406,12 @@ pub async fn collect(
plan: Arc<dyn ExecutionPlan>,
context: Arc<TaskContext>,
) -> Result<Vec<RecordBatch>> {
let stream = execute_stream(plan, context).await?;
let stream = execute_stream(plan, context)?;
common::collect(stream).await
}

/// Execute the [ExecutionPlan] and return a single stream of results
pub async fn execute_stream(
pub fn execute_stream(
plan: Arc<dyn ExecutionPlan>,
context: Arc<TaskContext>,
) -> Result<SendableRecordBatchStream> {
Expand All @@ -433,7 +433,7 @@ pub async fn collect_partitioned(
plan: Arc<dyn ExecutionPlan>,
context: Arc<TaskContext>,
) -> Result<Vec<Vec<RecordBatch>>> {
let streams = execute_stream_partitioned(plan, context).await?;
let streams = execute_stream_partitioned(plan, context)?;
let mut batches = Vec::with_capacity(streams.len());
for stream in streams {
batches.push(common::collect(stream).await?);
Expand All @@ -442,7 +442,7 @@ pub async fn collect_partitioned(
}

/// Execute the [ExecutionPlan] and return a vec with one stream per output partition
pub async fn execute_stream_partitioned(
pub fn execute_stream_partitioned(
plan: Arc<dyn ExecutionPlan>,
context: Arc<TaskContext>,
) -> Result<Vec<SendableRecordBatchStream>> {
Expand Down