risingwave_batch/execution/
grpc_exchange.rs

1// Copyright 2022 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::fmt::{Debug, Formatter};
16
17use await_tree::InstrumentAwait;
18use futures::StreamExt;
19use risingwave_common::array::DataChunk;
20use risingwave_expr::expr_context::capture_expr_context;
21use risingwave_pb::batch_plan::TaskOutputId;
22use risingwave_pb::batch_plan::exchange_source::LocalExecutePlan::{self, Plan};
23use risingwave_pb::task_service::{ExecuteRequest, GetDataResponse};
24use risingwave_rpc_client::ComputeClient;
25use risingwave_rpc_client::error::RpcError;
26use tonic::Streaming;
27
28use crate::error::Result;
29use crate::exchange_source::ExchangeSource;
30use crate::task::TaskId;
31
32/// Use grpc client as the source.
33pub struct GrpcExchangeSource {
34    stream: Streaming<GetDataResponse>,
35
36    task_output_id: TaskOutputId,
37
38    take_data_span: await_tree::Span,
39}
40
41impl GrpcExchangeSource {
42    pub async fn create(
43        client: ComputeClient,
44        task_output_id: TaskOutputId,
45        local_execute_plan: Option<LocalExecutePlan>,
46    ) -> Result<Self> {
47        let task_id = task_output_id.get_task_id()?.clone();
48        let take_data_span = await_tree::span!(
49            "grpc_exchange_take_data (query {} stage {} task {} output {})",
50            task_id.query_id,
51            task_id.stage_id,
52            task_id.task_id,
53            task_output_id.output_id
54        );
55        let stream = match local_execute_plan {
56            // When in the local execution mode, `GrpcExchangeSource` would send out
57            // `ExecuteRequest` and get the data chunks back in a single RPC.
58            Some(local_execute_plan) => {
59                let plan = try_match_expand!(local_execute_plan, Plan)?;
60                let execute_request = ExecuteRequest {
61                    task_id: Some(task_id),
62                    plan: plan.plan,
63                    tracing_context: plan.tracing_context,
64                    expr_context: Some(capture_expr_context()?),
65                };
66                client.execute(execute_request).await?
67            }
68            None => client.get_data(task_output_id.clone()).await?,
69        };
70        let source = Self {
71            stream,
72            task_output_id,
73            take_data_span,
74        };
75        Ok(source)
76    }
77}
78
79impl Debug for GrpcExchangeSource {
80    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
81        f.debug_struct("GrpcExchangeSource")
82            .field("task_output_id", &self.task_output_id)
83            .finish()
84    }
85}
86
87impl ExchangeSource for GrpcExchangeSource {
88    async fn take_data(&mut self) -> Result<Option<DataChunk>> {
89        let res = match self
90            .stream
91            .next()
92            .instrument_await(self.take_data_span.clone())
93            .await
94        {
95            None => {
96                return Ok(None);
97            }
98            Some(r) => r,
99        };
100        let task_data = res.map_err(RpcError::from_batch_status)?;
101        let data = DataChunk::from_protobuf(task_data.get_record_batch()?)?.compact_vis();
102        trace!(
103            "Receiver taskOutput = {:?}, data = {:?}",
104            self.task_output_id, data
105        );
106
107        Ok(Some(data))
108    }
109
110    fn get_task_id(&self) -> TaskId {
111        TaskId::from(self.task_output_id.get_task_id().unwrap())
112    }
113}
risingwave_batch/execution/grpc_exchange.rs

risingwave_batch/execution/
grpc_exchange.rs
