Split a large `NativeResponse` into multiple encoded frames that each fit within `MAX_FRAME_SIZE`. Intermediate frames use `Partial` status; the last frame uses the original status. Only responses with `rows` data are split. Error or auth responses that somehow exceed the frame limit are returned as-is (best effort).
(
response: NativeResponse,
format: codec::FrameFormat,
)
| 13 | /// Only responses with `rows` data are split. Error or auth responses |
| 14 | /// that somehow exceed the frame limit are returned as-is (best effort). |
| 15 | pub(super) fn chunk_large_response( |
| 16 | response: NativeResponse, |
| 17 | format: codec::FrameFormat, |
| 18 | ) -> crate::Result<Vec<Vec<u8>>> { |
| 19 | let rows = match response.rows { |
| 20 | Some(ref rows) if !rows.is_empty() => rows, |
| 21 | _ => { |
| 22 | // No rows to split — send as-is (shouldn't happen, but safe). |
| 23 | return Ok(vec![codec::encode_response(&response, format)?]); |
| 24 | } |
| 25 | }; |
| 26 | |
| 27 | // Estimate how many rows per chunk: target ~12 MiB per frame (75% of max) |
| 28 | // to leave headroom for envelope overhead. |
| 29 | let target_size = (MAX_FRAME_SIZE as usize) * 3 / 4; |
| 30 | let total_rows = rows.len(); |
| 31 | |
| 32 | // Estimate per-row size from a sample of the first rows. |
| 33 | let sample_resp = NativeResponse { |
| 34 | seq: response.seq, |
| 35 | status: ResponseStatus::Ok, |
| 36 | columns: response.columns.clone(), |
| 37 | rows: Some(rows[..total_rows.min(100)].to_vec()), |
| 38 | rows_affected: None, |
| 39 | watermark_lsn: response.watermark_lsn, |
| 40 | error: None, |
| 41 | auth: None, |
| 42 | warnings: Vec::new(), |
| 43 | }; |
| 44 | let sample_bytes = codec::encode_response(&sample_resp, format)?; |
| 45 | let sample_count = total_rows.min(100); |
| 46 | let per_row_estimate = sample_bytes.len().checked_div(sample_count).unwrap_or(256); |
| 47 | |
| 48 | let rows_per_chunk = target_size |
| 49 | .checked_div(per_row_estimate) |
| 50 | .map(|v| v.max(1)) |
| 51 | .unwrap_or(1000); |
| 52 | |
| 53 | let mut frames = Vec::new(); |
| 54 | let chunks: Vec<_> = rows.chunks(rows_per_chunk).collect(); |
| 55 | let last_idx = chunks.len().saturating_sub(1); |
| 56 | |
| 57 | for (i, chunk) in chunks.iter().enumerate() { |
| 58 | let is_last = i == last_idx; |
| 59 | let frame_resp = NativeResponse { |
| 60 | seq: response.seq, |
| 61 | status: if is_last { |
| 62 | response.status |
| 63 | } else { |
| 64 | ResponseStatus::Partial |
| 65 | }, |
| 66 | columns: if i == 0 { |
| 67 | response.columns.clone() |
| 68 | } else { |
| 69 | None |
| 70 | }, |
| 71 | rows: Some(chunk.to_vec()), |
| 72 | rows_affected: if is_last { |