async fn copy_from(
        &mut self,
        id: GlobalId,
        columns: Vec<usize>,
        params: CopyParams,
        row_desc: RelationDesc,
    ) -> Result<State, io::Error> {
        if !matches!(params.format, CopyFormat::Text | CopyFormat::Csv) {
            return self
                .error(ErrorResponse::error(
                    SqlState::FEATURE_NOT_SUPPORTED,
                    format!("COPY FROM format {:?} not supported", params.format),
                ))
                .await;
        }

        // Ensure params are valid here so as to error before waiting to receive
        // any data from the client.
        let params: CopyFormatParams = match params.try_into() {
            Ok(params) => params,
            Err(e) => {
                return self.error(e).await;
            }
        };

        let typ = row_desc.typ();
        let column_formats = vec![pgrepr::Format::Text; typ.column_types.len()];
        self.conn
            .send(BackendMessage::CopyInResponse {
                overall_format: pgrepr::Format::Text,
                column_formats,
            })
            .await?;
        self.conn.flush().await?;

        let mut data = Vec::new();
        let mut next_state = State::Ready;
        loop {
            let message = self.conn.recv().await?;
            match message {
                Some(FrontendMessage::CopyData(buf)) => data.extend(buf),
                Some(FrontendMessage::CopyDone) => break,
                Some(FrontendMessage::CopyFail(err)) => {
                    return self
                        .error(ErrorResponse::error(
                            SqlState::QUERY_CANCELED,
                            format!("COPY from stdin failed: {}", err),
                        ))
                        .await
                }
                Some(FrontendMessage::Flush) | Some(FrontendMessage::Sync) => {}
                Some(_) => {
                    return self
                        .error(ErrorResponse::error(
                            SqlState::PROTOCOL_VIOLATION,
                            "unexpected message type during COPY from stdin",
                        ))
                        .await
                }
                _ => {
                    next_state = State::Done;
                    break;
                }
            }
        }

        let column_types = typ
            .column_types
            .iter()
            .map(|x| &x.scalar_type)
            .map(pgrepr::Type::from)
            .collect::<Vec<pgrepr::Type>>();

        if let State::Ready = next_state {
            let rows = match decode_copy_format(&data, &column_types, params) {
                Ok(rows) => rows,
                Err(e) => {
                    return self
                        .error(ErrorResponse::error(
                            SqlState::BAD_COPY_FILE_FORMAT,
                            format!("{}", e),
                        ))
                        .await
                }
            };

            let count = rows.len();

            if let Err(e) = self.coord_client.insert_rows(id, columns, rows).await {
                return self
                    .error(ErrorResponse::from_coord(Severity::Error, e))
                    .await;
            }

            let tag = format!("COPY {}", count);
            self.conn
                .send(BackendMessage::CommandComplete { tag })
                .await?;
        }

        Ok(next_state)
    }
/// Handles the copy-in mode of the postgres protocol from transferring
/// data to the server.
Handles the copy-in mode of the postgres protocol from transferring
data to the server.
Handles the copy - in mode of the postgres protocol from transferring data to the server .
async fn copy_from(
        &mut self,
        id: GlobalId,
        columns: Vec<usize>,
        params: CopyParams,
        row_desc: RelationDesc,
    ) -> Result<State, io::Error> {
        if !matches!(params.format, CopyFormat::Text | CopyFormat::Csv) {
            return self
                .error(ErrorResponse::error(
                    SqlState::FEATURE_NOT_SUPPORTED,
                    format!("COPY FROM format {:?} not supported", params.format),
                ))
                .await;
        }
        // Ensure params are valid here so as to error before waiting to receive
        // any data from the client.
        let params: CopyFormatParams = match params.try_into() {
            Ok(params) => params,
            Err(e) => {
                return self.error(e).await;
            }
        };
        let typ = row_desc.typ();
        let column_formats = vec![pgrepr::Format::Text; typ.column_types.len()];
        self.conn
            .send(BackendMessage::CopyInResponse {
                overall_format: pgrepr::Format::Text,
                column_formats,
            })
            .await?;
        self.conn.flush().await?;
        let mut data = Vec::new();
        let mut next_state = State::Ready;
        loop {
            let message = self.conn.recv().await?;
            match message {
                Some(FrontendMessage::CopyData(buf)) => data.extend(buf),
                Some(FrontendMessage::CopyDone) => break,
                Some(FrontendMessage::CopyFail(err)) => {
                    return self
                        .error(ErrorResponse::error(
                            SqlState::QUERY_CANCELED,
                            format!("COPY from stdin failed: {}", err),
                        ))
                        .await
                }
                Some(FrontendMessage::Flush) | Some(FrontendMessage::Sync) => {}
                Some(_) => {
                    return self
                        .error(ErrorResponse::error(
                            SqlState::PROTOCOL_VIOLATION,
                            "unexpected message type during COPY from stdin",
                        ))
                        .await
                }
                _ => {
                    next_state = State::Done;
                    break;
                }
            }
        }
        let column_types = typ
            .column_types
            .iter()
            .map(|x| &x.scalar_type)
            .map(pgrepr::Type::from)
            .collect::<Vec<pgrepr::Type>>();
        if let State::Ready = next_state {
            let rows = match decode_copy_format(&data, &column_types, params) {
                Ok(rows) => rows,
                Err(e) => {
                    return self
                        .error(ErrorResponse::error(
                            SqlState::BAD_COPY_FILE_FORMAT,
                            format!("{}", e),
                        ))
                        .await
                }
            };
            let count = rows.len();
            if let Err(e) = self.coord_client.insert_rows(id, columns, rows).await {
                return self
                    .error(ErrorResponse::from_coord(Severity::Error, e))
                    .await;
            }
            let tag = format!("COPY {}", count);
            self.conn
                .send(BackendMessage::CommandComplete { tag })
                .await?;
        }
        Ok(next_state)
    }
async fn copy_from ( & mut self , id : GlobalId , columns : Vec < usize > , params : CopyParams , row_desc : RelationDesc , ) -> Result < State , io :: Error > { if ! matches ! ( params format CopyFormat Text CopyFormat Csv ) { return self . error ( ErrorResponse :: error ( SqlState :: FEATURE_NOT_SUPPORTED , format ! ( " " params format ) , ) ) . await ; } // Ensure params are valid here so as to error before waiting to receive // any data from the client. let params : CopyFormatParams = match params . try_into ( ) { Ok ( params ) => params , Err ( e ) => { return self . error ( e ) . await ; } } ; let typ = row_desc . typ ( ) ; let column_formats = vec ! [ pgrepr Format Text typ column_types len ( ) ] ; self . conn . send ( BackendMessage :: CopyInResponse { overall_format : pgrepr :: Format :: Text , column_formats , } ) . await ? ; self . conn . flush ( ) . await ? ; let mut data = Vec :: new ( ) ; let mut next_state = State :: Ready ; loop { let message = self . conn . recv ( ) . await ? ; match message { Some ( FrontendMessage :: CopyData ( buf ) ) => data . extend ( buf ) , Some ( FrontendMessage :: CopyDone ) => break , Some ( FrontendMessage :: CopyFail ( err ) ) => { return self . error ( ErrorResponse :: error ( SqlState :: QUERY_CANCELED , format ! ( " " err ) , ) ) . await } Some ( FrontendMessage :: Flush ) | Some ( FrontendMessage :: Sync ) => { } Some ( _ ) => { return self . error ( ErrorResponse :: error ( SqlState :: PROTOCOL_VIOLATION , " " , ) ) . await } _ => { next_state = State :: Done ; break ; } } } let column_types = typ . column_types . iter ( ) . map ( | x | & x . scalar_type ) . map ( pgrepr :: Type :: from ) . collect :: < Vec < pgrepr :: Type > > ( ) ; if let State :: Ready = next_state { let rows = match decode_copy_format ( & data , & column_types , params ) { Ok ( rows ) => rows , Err ( e ) => { return self . error ( ErrorResponse :: error ( SqlState :: BAD_COPY_FILE_FORMAT , format ! ( " " e ) , ) ) . await } } ; let count = rows . len ( ) ; if let Err ( e ) = self . coord_client . insert_rows ( id , columns , rows ) . await { return self . error ( ErrorResponse :: from_coord ( Severity :: Error , e ) ) . await ; } let tag = format ! ( " " count ) ; self . conn . send ( BackendMessage :: CommandComplete { tag } ) . await ? ; } Ok ( next_state ) }
Handles the copy-in mode of the postgres protocol from transferring
data to the server.
Handles the copy - in mode of the postgres protocol from transferring data to the server .
&mut self
id
GlobalId
columns
Vec<usize>
params
CopyParams
row_desc
RelationDesc
&mut self
id
GlobalId
columns
Vec<usize>
params
CopyParams
row_desc
RelationDesc