Method query_internal

src/pglite/client.rs:303–384

(
        &mut self,
        sql: &str,
        params: &[Value],
        options: Option<&QueryOptions>,
    )

    }

    fn query_internal(
        &mut self,
        sql: &str,
        params: &[Value],
        options: Option<&QueryOptions>,
    ) -> Result<Results> {
        let default_options = QueryOptions::default();
        let query_opts = options.unwrap_or(&default_options);

        self.handle_blob_input(query_opts.blob.as_ref())?;

        let params_snapshot: Vec<Value> = params.to_vec();
        let options_snapshot = options.cloned();
        let mut collected_messages: Vec<BackendMessage> = Vec::new();

        let mut exec_opts = ExecProtocolOptions::no_sync();
        exec_opts.on_notice = query_opts.on_notice.clone();
        exec_opts.data_transfer_container = query_opts.data_transfer_container;

        let result: Result<()> = (|| {
            let param_types = if query_opts.param_types.is_empty() {
                &[] as &[i32]
            } else {
                &query_opts.param_types
            };

            let mut messages = {
                let _phase = timing::phase("client.query.parse_describe");
                self.parse_and_describe(sql, param_types, exec_opts.clone())?
            };
            let mut data_type_ids = parse_describe_statement_results(&messages);
            if self.ensure_array_types_for_bind_values(params, &data_type_ids, query_opts)? {
                messages = {
                    let _phase = timing::phase("client.query.parse_describe_after_array_register");
                    self.parse_and_describe(sql, param_types, exec_opts.clone())?
                };
                data_type_ids = parse_describe_statement_results(&messages);
            }
            collected_messages.extend(messages);
            let bind_values = {
                let _phase = timing::phase("client.query.prepare_bind_values");
                self.prepare_bind_values(params, &data_type_ids, query_opts)?
            };
            let bind_config = BindConfig {
                values: bind_values,
                ..Default::default()
            };
            let execute_batch = {
                let _phase = timing::phase("client.query.serialize_execute");
                let mut execute_batch = Vec::new();
                execute_batch.extend(Serialize::bind(&bind_config));
                execute_batch.extend(Serialize::describe(&PortalTarget::new('P', None)));
                execute_batch.extend(Serialize::execute(None));
                execute_batch.extend(Serialize::sync());
                execute_batch
            };
            let ExecProtocolResult { messages, .. } = {
                let _phase = timing::phase("client.query.execute_roundtrip");