Method copy_from_inner

src/pgwire/src/protocol.rs:2787–3079

(
        &mut self,
        target_id: CatalogItemId,
        target_name: String,
        columns: Vec<ColumnIndex>,
        params: CopyFormatParams<'static>,
        row_desc: RelationDesc,
      

    }

    async fn copy_from_inner(
        &mut self,
        target_id: CatalogItemId,
        target_name: String,
        columns: Vec<ColumnIndex>,
        params: CopyFormatParams<'static>,
        row_desc: RelationDesc,
        ctx_extra: &mut ExecuteContextGuard,
    ) -> Result<State, io::Error> {
        let typ = row_desc.typ();
        let column_formats = vec![Format::Text; typ.column_types.len()];
        self.send(BackendMessage::CopyInResponse {
            overall_format: Format::Text,
            column_formats,
        })
        .await?;
        self.conn.flush().await?;

        // Set up the parallel streaming batch builders in the coordinator.
        let writer = match self
            .adapter_client
            .start_copy_from_stdin(
                target_id,
                target_name.clone(),
                columns.clone(),
                row_desc.clone(),
                params.clone(),
            )
            .await
        {
            Ok(writer) => writer,
            Err(e) => {
                // Drain remaining CopyData/CopyDone/CopyFail messages from the
                // socket. Since CopyInResponse was already sent, the client may
                // have pipelined copy data that we must consume before returning
                // the error, otherwise they'd be misinterpreted as top-level
                // protocol messages and cause a deadlock.
                loop {
                    match self.conn.recv().await? {
                        Some(FrontendMessage::CopyData(_)) => {}
                        Some(FrontendMessage::CopyDone) | Some(FrontendMessage::CopyFail(_)) => {
                            break;
                        }
                        Some(FrontendMessage::Flush) | Some(FrontendMessage::Sync) => {}
                        Some(_) => break,
                        None => return Ok(State::Done),
                    }
                }
                self.adapter_client.retire_execute(
                    std::mem::take(ctx_extra),
                    StatementEndedExecutionReason::Errored {
                        error: e.to_string(),
                    },
                );
                return self
                    .send_error_and_get_state(e.into_response(Severity::Error))
                    .await;
            }