Method copy_rows

src/pgwire/src/protocol.rs:2578–2732

(
        &mut self,
        format: CopyFormat,
        row_desc: RelationDesc,
        mut stream: RecordFirstRowStream,
    )

    #[mz_ore::instrument(level = "debug")]
    async fn copy_rows(
        &mut self,
        format: CopyFormat,
        row_desc: RelationDesc,
        mut stream: RecordFirstRowStream,
    ) -> Result<(State, SendRowsEndedReason), io::Error> {
        let (row_format, encode_format) = match format {
            CopyFormat::Text => (
                CopyFormatParams::Text(CopyTextFormatParams::default()),
                Format::Text,
            ),
            CopyFormat::Binary => (CopyFormatParams::Binary, Format::Binary),
            CopyFormat::Csv => (
                CopyFormatParams::Csv(CopyCsvFormatParams::default()),
                Format::Text,
            ),
            CopyFormat::Parquet => {
                let text = "Parquet format is not supported".to_string();
                return self
                    .send_error_and_get_state(ErrorResponse::error(
                        SqlState::INTERNAL_ERROR,
                        text.clone(),
                    ))
                    .await
                    .map(|state| (state, SendRowsEndedReason::Errored { error: text }));
            }
        };

        // Binary encoding is not implemented for some types (e.g., list, map,
        // and aclitem). Unlike the extended query protocol's Bind handler, COPY
        // does not validate this when binding the portal: the portal's result
        // formats describe the `CopyData` wrapper, not the COPY format itself,
        // so the Bind handler explicitly skips `COPY TO` statements. We must
        // therefore check here, before streaming any rows, otherwise
        // `encode_binary` would panic mid-stream (SQL-323).
        if let CopyFormat::Binary = format {
            if let Some(msg) = row_desc
                .iter_types()
                .find_map(|ty| mz_pgrepr::Value::binary_encoding_error(&ty.scalar_type).err())
            {
                return self
                    .send_error_and_get_state(ErrorResponse::error(
                        SqlState::UNDEFINED_FUNCTION,
                        msg,
                    ))
                    .await
                    .map(|state| {
                        (
                            state,
                            SendRowsEndedReason::Errored {
                                error: msg.to_string(),
                            },
                        )
                    });
            }
        }

        let encode_fn = |row: &RowRef, typ: &SqlRelationType, out: &mut Vec<u8>| {