(
&mut self,
row_desc: RelationDesc,
portal_name: String,
mut rows: InProgressRows,
max_rows: ExecuteCount,
get_response: GetResponse,
fetch_por
| 2303 | // TODO(guswynn): figure out how to get it to compile without skip_all |
| 2304 | #[mz_ore::instrument(level = "debug")] |
| 2305 | async fn send_rows( |
| 2306 | &mut self, |
| 2307 | row_desc: RelationDesc, |
| 2308 | portal_name: String, |
| 2309 | mut rows: InProgressRows, |
| 2310 | max_rows: ExecuteCount, |
| 2311 | get_response: GetResponse, |
| 2312 | fetch_portal_name: Option<String>, |
| 2313 | timeout: ExecuteTimeout, |
| 2314 | ) -> Result<(State, SendRowsEndedReason), io::Error> { |
| 2315 | // If this portal is being executed from a FETCH then we need to use the result |
| 2316 | // format type of the outer portal. |
| 2317 | let result_format_portal_name: &str = if let Some(ref name) = fetch_portal_name { |
| 2318 | name |
| 2319 | } else { |
| 2320 | &portal_name |
| 2321 | }; |
| 2322 | let result_formats = self |
| 2323 | .adapter_client |
| 2324 | .session() |
| 2325 | .get_portal_unverified(result_format_portal_name) |
| 2326 | .expect("valid fetch portal name for send rows") |
| 2327 | .result_formats |
| 2328 | .clone(); |
| 2329 | |
| 2330 | let (mut wait_once, mut deadline) = match timeout { |
| 2331 | ExecuteTimeout::None => (false, None), |
| 2332 | ExecuteTimeout::Seconds(t) => ( |
| 2333 | false, |
| 2334 | Some(tokio::time::Instant::now() + tokio::time::Duration::from_secs_f64(t)), |
| 2335 | ), |
| 2336 | ExecuteTimeout::WaitOnce => (true, None), |
| 2337 | }; |
| 2338 | |
| 2339 | // Sanity check that the various `RelationDesc`s match up. |
| 2340 | { |
| 2341 | let portal_name_desc = &self |
| 2342 | .adapter_client |
| 2343 | .session() |
| 2344 | .get_portal_unverified(portal_name.as_str()) |
| 2345 | .expect("portal should exist") |
| 2346 | .desc |
| 2347 | .relation_desc; |
| 2348 | if let Some(portal_name_desc) = portal_name_desc { |
| 2349 | soft_assert_eq_or_log!(portal_name_desc, &row_desc); |
| 2350 | } |
| 2351 | if let Some(fetch_portal_name) = &fetch_portal_name { |
| 2352 | let fetch_portal_desc = &self |
| 2353 | .adapter_client |
| 2354 | .session() |
| 2355 | .get_portal_unverified(fetch_portal_name) |
| 2356 | .expect("portal should exist") |
| 2357 | .desc |
| 2358 | .relation_desc; |
| 2359 | if let Some(fetch_portal_desc) = fetch_portal_desc { |
| 2360 | soft_assert_eq_or_log!(fetch_portal_desc, &row_desc); |
| 2361 | } |
| 2362 | } |
no test coverage detected