From 63e278b9f293fc8c2f15092cef8d8da3dcb153b1 Mon Sep 17 00:00:00 2001 From: Steven Fackler Date: Sat, 15 Aug 2015 21:39:46 -0700 Subject: [PATCH] Pull Statement into its own module --- src/lib.rs | 440 ++++------------------------------------------------ src/rows.rs | 17 +- src/stmt.rs | 428 ++++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 464 insertions(+), 421 deletions(-) create mode 100644 src/stmt.rs diff --git a/src/lib.rs b/src/lib.rs index 985e1964..e104e468 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -70,10 +70,12 @@ use std::result; #[cfg(feature = "unix_socket")] use std::path::PathBuf; +pub use stmt::{Statement, Column}; + use error::{Error, ConnectError, SqlState, DbError}; use types::{ToSql, FromSql}; use io::{StreamWrapper, NegotiateSsl}; -use types::{IsNull, Kind, Type, SessionInfo, Oid, Other, ReadWithInfo}; +use types::{IsNull, Kind, Type, SessionInfo, Oid, Other}; use message::BackendMessage::*; use message::FrontendMessage::*; use message::{FrontendMessage, BackendMessage, RowDescriptionEntry}; @@ -84,15 +86,16 @@ use rows::{Rows, LazyRows}; #[macro_use] mod macros; -pub mod error; -pub mod io; +mod md5; mod message; mod priv_io; +mod stmt; mod url; mod util; -pub mod types; +pub mod error; +pub mod io; pub mod rows; -mod md5; +pub mod types; const TYPEINFO_QUERY: &'static str = "t"; @@ -663,10 +666,7 @@ impl InnerConnection { let mut columns = vec![]; for RowDescriptionEntry { name, type_oid, .. } in raw_columns { - columns.push(Column { - name: name, - type_: try!(self.get_type(type_oid)), - }); + columns.push(Column::new(name, try!(self.get_type(type_oid)))); } Ok((param_types, columns)) @@ -681,14 +681,7 @@ impl InnerConnection { fn prepare<'a>(&mut self, query: &str, conn: &'a Connection) -> Result> { let stmt_name = self.make_stmt_name(); let (param_types, columns) = try!(self.raw_prepare(&stmt_name, query)); - Ok(Statement { - conn: conn, - name: stmt_name, - param_types: param_types, - columns: columns, - next_portal_id: Cell::new(0), - finished: false, - }) + Ok(Statement::new(conn, stmt_name, param_types, columns, Cell::new(0), false)) } fn prepare_cached<'a>(&mut self, query: &str, conn: &'a Connection) -> Result> { @@ -709,14 +702,7 @@ impl InnerConnection { } }; - Ok(Statement { - conn: conn, - name: name, - param_types: param_types, - columns: columns, - next_portal_id: Cell::new(0), - finished: true, // << ! - }) + Ok(Statement::new(conn, name, param_types, columns, Cell::new(0), true)) } fn close_statement(&mut self, name: &str, type_: u8) -> Result<()> { @@ -1082,14 +1068,7 @@ impl Connection { /// expected. pub fn execute(&self, query: &str, params: &[&ToSql]) -> Result { let (param_types, columns) = try!(self.conn.borrow_mut().raw_prepare("", query)); - let stmt = Statement { - conn: self, - name: "".to_owned(), - param_types: param_types, - columns: columns, - next_portal_id: Cell::new(0), - finished: true, // << !! - }; + let stmt = Statement::new(self, "".to_owned(), param_types, columns, Cell::new(0), true); stmt.execute(params) } @@ -1300,386 +1279,6 @@ impl<'conn> Transaction<'conn> { } } -/// A prepared statement. -pub struct Statement<'conn> { - conn: &'conn Connection, - name: String, - param_types: Vec, - columns: Vec, - next_portal_id: Cell, - finished: bool, -} - -impl<'a> fmt::Debug for Statement<'a> { - fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { - DebugStruct::new(fmt, "Statement") - .field("name", &self.name) - .field("parameter_types", &self.param_types) - .field("columns", &self.columns) - .finish() - } -} - -impl<'conn> Drop for Statement<'conn> { - fn drop(&mut self) { - let _ = self.finish_inner(); - } -} - -impl<'conn> Statement<'conn> { - fn finish_inner(&mut self) -> Result<()> { - if !self.finished { - self.finished = true; - let mut conn = self.conn.conn.borrow_mut(); - check_desync!(conn); - conn.close_statement(&self.name, b'S') - } else { - Ok(()) - } - } - - fn inner_execute(&self, portal_name: &str, row_limit: i32, params: &[&ToSql]) -> Result<()> { - let mut conn = self.conn.conn.borrow_mut(); - assert!(self.param_types().len() == params.len(), - "expected {} parameters but got {}", - self.param_types.len(), - params.len()); - debug!("executing statement {} with parameters: {:?}", self.name, params); - let mut values = vec![]; - for (param, ty) in params.iter().zip(self.param_types.iter()) { - let mut buf = vec![]; - match try!(param.to_sql_checked(ty, &mut buf, &SessionInfo::new(&*conn))) { - IsNull::Yes => values.push(None), - IsNull::No => values.push(Some(buf)), - } - }; - - try!(conn.write_messages(&[ - Bind { - portal: portal_name, - statement: &self.name, - formats: &[1], - values: &values, - result_formats: &[1] - }, - Execute { - portal: portal_name, - max_rows: row_limit - }, - Sync])); - - match try!(conn.read_message()) { - BindComplete => Ok(()), - ErrorResponse { fields } => { - try!(conn.wait_for_ready()); - DbError::new(fields) - } - _ => { - conn.desynchronized = true; - Err(Error::IoError(bad_response())) - } - } - } - - fn inner_query<'a>(&'a self, portal_name: &str, row_limit: i32, params: &[&ToSql]) - -> Result<(VecDeque>>>, bool)> { - try!(self.inner_execute(portal_name, row_limit, params)); - - let mut buf = VecDeque::new(); - let more_rows = try!(read_rows(&mut self.conn.conn.borrow_mut(), &mut buf)); - Ok((buf, more_rows)) - } - - /// Returns a slice containing the expected parameter types. - pub fn param_types(&self) -> &[Type] { - &self.param_types - } - - /// Returns a slice describing the columns of the result of the query. - pub fn columns(&self) -> &[Column] { - &self.columns - } - - /// Executes the prepared statement, returning the number of rows modified. - /// - /// If the statement does not modify any rows (e.g. SELECT), 0 is returned. - /// - /// ## Panics - /// - /// Panics if the number of parameters provided does not match the number - /// expected. - /// - /// ## Example - /// - /// ```rust,no_run - /// # use postgres::{Connection, SslMode}; - /// # let conn = Connection::connect("", &SslMode::None).unwrap(); - /// # let bar = 1i32; - /// # let baz = true; - /// let stmt = conn.prepare("UPDATE foo SET bar = $1 WHERE baz = $2").unwrap(); - /// match stmt.execute(&[&bar, &baz]) { - /// Ok(count) => println!("{} row(s) updated", count), - /// Err(err) => println!("Error executing query: {:?}", err) - /// } - /// ``` - pub fn execute(&self, params: &[&ToSql]) -> Result { - check_desync!(self.conn); - try!(self.inner_execute("", 0, params)); - - let mut conn = self.conn.conn.borrow_mut(); - let num; - loop { - match try!(conn.read_message()) { - DataRow { .. } => {} - ErrorResponse { fields } => { - try!(conn.wait_for_ready()); - return DbError::new(fields); - } - CommandComplete { tag } => { - num = util::parse_update_count(tag); - break; - } - EmptyQueryResponse => { - num = 0; - break; - } - CopyInResponse { .. } => { - try!(conn.write_messages(&[ - CopyFail { - message: "COPY queries cannot be directly executed", - }, - Sync])); - } - CopyOutResponse { .. } => { - loop { - match try!(conn.read_message()) { - BCopyDone => break, - ErrorResponse { fields } => { - try!(conn.wait_for_ready()); - return DbError::new(fields); - } - _ => {} - } - } - num = 0; - break; - } - _ => { - conn.desynchronized = true; - return Err(Error::IoError(bad_response())); - } - } - } - try!(conn.wait_for_ready()); - - Ok(num) - } - - /// Executes the prepared statement, returning the resulting rows. - /// - /// ## Panics - /// - /// Panics if the number of parameters provided does not match the number - /// expected. - /// - /// ## Example - /// - /// ```rust,no_run - /// # use postgres::{Connection, SslMode}; - /// # let conn = Connection::connect("", &SslMode::None).unwrap(); - /// let stmt = conn.prepare("SELECT foo FROM bar WHERE baz = $1").unwrap(); - /// # let baz = true; - /// let rows = match stmt.query(&[&baz]) { - /// Ok(rows) => rows, - /// Err(err) => panic!("Error running query: {:?}", err) - /// }; - /// for row in &rows { - /// let foo: i32 = row.get("foo"); - /// println!("foo: {}", foo); - /// } - /// ``` - pub fn query<'a>(&'a self, params: &[&ToSql]) -> Result> { - check_desync!(self.conn); - self.inner_query("", 0, params).map(|(buf, _)| { - Rows::new(self, buf.into_iter().collect()) - }) - } - - /// Executes the prepared statement, returning a lazily loaded iterator - /// over the resulting rows. - /// - /// No more than `row_limit` rows will be stored in memory at a time. Rows - /// will be pulled from the database in batches of `row_limit` as needed. - /// If `row_limit` is less than or equal to 0, `lazy_query` is equivalent - /// to `query`. - /// - /// This can only be called inside of a transaction, and the `Transaction` - /// object representing the active transaction must be passed to - /// `lazy_query`. - /// - /// ## Panics - /// - /// Panics if the provided `Transaction` is not associated with the same - /// `Connection` as this `Statement`, if the `Transaction` is not - /// active, or if the number of parameters provided does not match the - /// number of parameters expected. - pub fn lazy_query<'trans, 'stmt>(&'stmt self, - trans: &'trans Transaction, - params: &[&ToSql], - row_limit: i32) - -> Result> { - assert!(self.conn as *const _ == trans.conn as *const _, - "the `Transaction` passed to `lazy_query` must be associated with the same \ - `Connection` as the `Statement`"); - let conn = self.conn.conn.borrow(); - check_desync!(conn); - assert!(conn.trans_depth == trans.depth, - "`lazy_query` must be passed the active transaction"); - drop(conn); - - let id = self.next_portal_id.get(); - self.next_portal_id.set(id + 1); - let portal_name = format!("{}p{}", self.name, id); - - self.inner_query(&portal_name, row_limit, params).map(move |(data, more_rows)| { - LazyRows::new(self, data, portal_name, row_limit, more_rows, false, trans) - }) - } - - /// Executes a `COPY FROM STDIN` statement, returning the number of rows - /// added. - /// - /// The contents of the provided reader are passed to the Postgres server - /// verbatim; it is the caller's responsibility to ensure it uses the - /// proper format. See the - /// [Postgres documentation](http://www.postgresql.org/docs/9.4/static/sql-copy.html) - /// for details. - /// - /// If the statement is not a `COPY FROM STDIN` statement it will still be - /// executed and this method will return an error. - /// - /// # Examples - /// - /// ```rust,no_run - /// # use postgres::{Connection, SslMode}; - /// # let conn = Connection::connect("", &SslMode::None).unwrap(); - /// conn.batch_execute("CREATE TABLE people (id INT PRIMARY KEY, name VARCHAR)").unwrap(); - /// let stmt = conn.prepare("COPY people FROM STDIN").unwrap(); - /// stmt.copy_in(&[], &mut "1\tjohn\n2\tjane\n".as_bytes()).unwrap(); - /// ``` - pub fn copy_in(&self, params: &[&ToSql], r: &mut R) -> Result { - try!(self.inner_execute("", 0, params)); - let mut conn = self.conn.conn.borrow_mut(); - - match try!(conn.read_message()) { - CopyInResponse { .. } => {} - _ => { - loop { - match try!(conn.read_message()) { - ReadyForQuery { .. } => { - return Err(Error::IoError(std_io::Error::new( - std_io::ErrorKind::InvalidInput, - "called `copy_in` on a non-`COPY FROM STDIN` statement"))); - } - _ => {} - } - } - } - } - - let mut buf = [0; 16 * 1024]; - loop { - match fill_copy_buf(&mut buf, r, &SessionInfo::new(&conn)) { - Ok(0) => break, - Ok(len) => { - try_desync!(conn, conn.stream.write_message( - &CopyData { - data: &buf[..len], - })); - } - Err(err) => { - try!(conn.write_messages(&[ - CopyFail { - message: "", - }, - CopyDone, - Sync])); - match try!(conn.read_message()) { - ErrorResponse { .. } => { /* expected from the CopyFail */ } - _ => { - conn.desynchronized = true; - return Err(Error::IoError(bad_response())); - } - } - try!(conn.wait_for_ready()); - return Err(Error::IoError(err)); - } - } - } - - try!(conn.write_messages(&[CopyDone, Sync])); - - let num = match try!(conn.read_message()) { - CommandComplete { tag } => util::parse_update_count(tag), - ErrorResponse { fields } => { - try!(conn.wait_for_ready()); - return DbError::new(fields); - } - _ => { - conn.desynchronized = true; - return Err(Error::IoError(bad_response())); - } - }; - - try!(conn.wait_for_ready()); - Ok(num) - } - - /// Consumes the statement, clearing it from the Postgres session. - /// - /// If this statement was created via the `prepare_cached` method, `finish` - /// does nothing. - /// - /// Functionally identical to the `Drop` implementation of the - /// `Statement` except that it returns any error to the caller. - pub fn finish(mut self) -> Result<()> { - self.finish_inner() - } -} - -fn fill_copy_buf(buf: &mut [u8], r: &mut R, info: &SessionInfo) - -> std_io::Result { - let mut nread = 0; - while nread < buf.len() { - match r.read_with_info(&mut buf[nread..], info) { - Ok(0) => break, - Ok(n) => nread += n, - Err(ref e) if e.kind() == std_io::ErrorKind::Interrupted => {} - Err(e) => return Err(e), - } - } - Ok(nread) -} - -/// Information about a column of the result of a query. -#[derive(PartialEq, Eq, Clone, Debug)] -pub struct Column { - name: String, - type_: Type -} - -impl Column { - /// The name of the column. - pub fn name(&self) -> &str { - &self.name - } - - /// The type of the data in the column. - pub fn type_(&self) -> &Type { - &self.type_ - } -} - fn read_rows(conn: &mut InnerConnection, buf: &mut VecDeque>>>) -> Result { let more_rows; loop { @@ -1829,3 +1428,18 @@ trait LazyRowsNew<'trans, 'stmt> { trait SessionInfoNew<'a> { fn new(conn: &'a InnerConnection) -> SessionInfo<'a>; } + +trait StatementInternals<'conn> { + fn new(conn: &'conn Connection, + name: String, + param_types: Vec, + columns: Vec, + next_portal_id: Cell, + finished: bool) -> Statement<'conn>; + + fn conn(&self) -> &'conn Connection; +} + +trait ColumnNew { + fn new(name: String, type_: Type) -> Column; +} diff --git a/src/rows.rs b/src/rows.rs index 1d695d68..7e27af00 100644 --- a/src/rows.rs +++ b/src/rows.rs @@ -16,7 +16,8 @@ use {Statement, DbErrorNew, SessionInfoNew, RowsNew, - LazyRowsNew}; + LazyRowsNew, + StatementInternals}; use types::{FromSql, SessionInfo}; use error::Error; use message::FrontendMessage::*; @@ -204,11 +205,11 @@ impl<'a> Row<'a> { /// the return type is not compatible with the Postgres type. pub fn get_opt(&self, idx: I) -> Result where I: RowIndex, T: FromSql { let idx = try!(idx.idx(self.stmt).ok_or(Error::InvalidColumn)); - let ty = &self.stmt.columns[idx].type_; + let ty = self.stmt.columns()[idx].type_(); if !::accepts(ty) { return Err(Error::WrongType(ty.clone())); } - let conn = self.stmt.conn.conn.borrow(); + let conn = self.stmt.conn().conn.borrow(); FromSql::from_sql_nullable(ty, self.data[idx].as_ref().map(|e| &**e).as_mut(), &SessionInfo::new(&*conn)) } @@ -264,7 +265,7 @@ pub trait RowIndex { impl RowIndex for usize { #[inline] fn idx(&self, stmt: &Statement) -> Option { - if *self >= stmt.columns.len() { + if *self >= stmt.columns().len() { None } else { Some(*self) @@ -275,14 +276,14 @@ impl RowIndex for usize { impl<'a> RowIndex for &'a str { #[inline] fn idx(&self, stmt: &Statement) -> Option { - if let Some(idx) = stmt.columns().iter().position(|d| d.name == *self) { + if let Some(idx) = stmt.columns().iter().position(|d| d.name() == *self) { return Some(idx); }; // FIXME ASCII-only case insensitivity isn't really the right thing to // do. Postgres itself uses a dubious wrapper around tolower and JDBC // uses the US locale. - stmt.columns().iter().position(|d| d.name.eq_ignore_ascii_case(*self)) + stmt.columns().iter().position(|d| d.name().eq_ignore_ascii_case(*self)) } } @@ -338,13 +339,13 @@ impl<'a, 'b> fmt::Debug for LazyRows<'a, 'b> { impl<'trans, 'stmt> LazyRows<'trans, 'stmt> { fn finish_inner(&mut self) -> Result<()> { - let mut conn = self.stmt.conn.conn.borrow_mut(); + let mut conn = self.stmt.conn().conn.borrow_mut(); check_desync!(conn); conn.close_statement(&self.name, b'P') } fn execute(&mut self) -> Result<()> { - let mut conn = self.stmt.conn.conn.borrow_mut(); + let mut conn = self.stmt.conn().conn.borrow_mut(); try!(conn.write_messages(&[ Execute { diff --git a/src/stmt.rs b/src/stmt.rs new file mode 100644 index 00000000..13f2c593 --- /dev/null +++ b/src/stmt.rs @@ -0,0 +1,428 @@ +use debug_builders::DebugStruct; +use std::cell::Cell; +use std::collections::VecDeque; +use std::fmt; +use std::io; + +use error::{Error, DbError}; +use types::{ReadWithInfo, SessionInfo, Type, ToSql, IsNull}; +use message::FrontendMessage::*; +use message::BackendMessage::*; +use message::WriteMessage; +use util; +use rows::{Rows, LazyRows}; +use {read_rows, bad_response, Connection, Transaction, StatementInternals, Result, RowsNew}; +use {SessionInfoNew, LazyRowsNew, DbErrorNew, ColumnNew}; + +/// A prepared statement. +pub struct Statement<'conn> { + conn: &'conn Connection, + name: String, + param_types: Vec, + columns: Vec, + next_portal_id: Cell, + finished: bool, +} + +impl<'a> fmt::Debug for Statement<'a> { + fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { + DebugStruct::new(fmt, "Statement") + .field("name", &self.name) + .field("parameter_types", &self.param_types) + .field("columns", &self.columns) + .finish() + } +} + +impl<'conn> Drop for Statement<'conn> { + fn drop(&mut self) { + let _ = self.finish_inner(); + } +} + +impl<'conn> StatementInternals<'conn> for Statement<'conn> { + fn new(conn: &'conn Connection, + name: String, + param_types: Vec, + columns: Vec, + next_portal_id: Cell, + finished: bool) -> Statement<'conn> { + Statement { + conn: conn, + name: name, + param_types: param_types, + columns: columns, + next_portal_id: next_portal_id, + finished: finished, + } + } + + fn conn(&self) -> &'conn Connection { + self.conn + } +} + +impl<'conn> Statement<'conn> { + fn finish_inner(&mut self) -> Result<()> { + if !self.finished { + self.finished = true; + let mut conn = self.conn.conn.borrow_mut(); + check_desync!(conn); + conn.close_statement(&self.name, b'S') + } else { + Ok(()) + } + } + + fn inner_execute(&self, portal_name: &str, row_limit: i32, params: &[&ToSql]) -> Result<()> { + let mut conn = self.conn.conn.borrow_mut(); + assert!(self.param_types().len() == params.len(), + "expected {} parameters but got {}", + self.param_types.len(), + params.len()); + debug!("executing statement {} with parameters: {:?}", self.name, params); + let mut values = vec![]; + for (param, ty) in params.iter().zip(self.param_types.iter()) { + let mut buf = vec![]; + match try!(param.to_sql_checked(ty, &mut buf, &SessionInfo::new(&*conn))) { + IsNull::Yes => values.push(None), + IsNull::No => values.push(Some(buf)), + } + }; + + try!(conn.write_messages(&[ + Bind { + portal: portal_name, + statement: &self.name, + formats: &[1], + values: &values, + result_formats: &[1] + }, + Execute { + portal: portal_name, + max_rows: row_limit + }, + Sync])); + + match try!(conn.read_message()) { + BindComplete => Ok(()), + ErrorResponse { fields } => { + try!(conn.wait_for_ready()); + DbError::new(fields) + } + _ => { + conn.desynchronized = true; + Err(Error::IoError(bad_response())) + } + } + } + + fn inner_query<'a>(&'a self, portal_name: &str, row_limit: i32, params: &[&ToSql]) + -> Result<(VecDeque>>>, bool)> { + try!(self.inner_execute(portal_name, row_limit, params)); + + let mut buf = VecDeque::new(); + let more_rows = try!(read_rows(&mut self.conn.conn.borrow_mut(), &mut buf)); + Ok((buf, more_rows)) + } + + /// Returns a slice containing the expected parameter types. + pub fn param_types(&self) -> &[Type] { + &self.param_types + } + + /// Returns a slice describing the columns of the result of the query. + pub fn columns(&self) -> &[Column] { + &self.columns + } + + /// Executes the prepared statement, returning the number of rows modified. + /// + /// If the statement does not modify any rows (e.g. SELECT), 0 is returned. + /// + /// ## Panics + /// + /// Panics if the number of parameters provided does not match the number + /// expected. + /// + /// ## Example + /// + /// ```rust,no_run + /// # use postgres::{Connection, SslMode}; + /// # let conn = Connection::connect("", &SslMode::None).unwrap(); + /// # let bar = 1i32; + /// # let baz = true; + /// let stmt = conn.prepare("UPDATE foo SET bar = $1 WHERE baz = $2").unwrap(); + /// match stmt.execute(&[&bar, &baz]) { + /// Ok(count) => println!("{} row(s) updated", count), + /// Err(err) => println!("Error executing query: {:?}", err) + /// } + /// ``` + pub fn execute(&self, params: &[&ToSql]) -> Result { + check_desync!(self.conn); + try!(self.inner_execute("", 0, params)); + + let mut conn = self.conn.conn.borrow_mut(); + let num; + loop { + match try!(conn.read_message()) { + DataRow { .. } => {} + ErrorResponse { fields } => { + try!(conn.wait_for_ready()); + return DbError::new(fields); + } + CommandComplete { tag } => { + num = util::parse_update_count(tag); + break; + } + EmptyQueryResponse => { + num = 0; + break; + } + CopyInResponse { .. } => { + try!(conn.write_messages(&[ + CopyFail { + message: "COPY queries cannot be directly executed", + }, + Sync])); + } + CopyOutResponse { .. } => { + loop { + match try!(conn.read_message()) { + BCopyDone => break, + ErrorResponse { fields } => { + try!(conn.wait_for_ready()); + return DbError::new(fields); + } + _ => {} + } + } + num = 0; + break; + } + _ => { + conn.desynchronized = true; + return Err(Error::IoError(bad_response())); + } + } + } + try!(conn.wait_for_ready()); + + Ok(num) + } + + /// Executes the prepared statement, returning the resulting rows. + /// + /// ## Panics + /// + /// Panics if the number of parameters provided does not match the number + /// expected. + /// + /// ## Example + /// + /// ```rust,no_run + /// # use postgres::{Connection, SslMode}; + /// # let conn = Connection::connect("", &SslMode::None).unwrap(); + /// let stmt = conn.prepare("SELECT foo FROM bar WHERE baz = $1").unwrap(); + /// # let baz = true; + /// let rows = match stmt.query(&[&baz]) { + /// Ok(rows) => rows, + /// Err(err) => panic!("Error running query: {:?}", err) + /// }; + /// for row in &rows { + /// let foo: i32 = row.get("foo"); + /// println!("foo: {}", foo); + /// } + /// ``` + pub fn query<'a>(&'a self, params: &[&ToSql]) -> Result> { + check_desync!(self.conn); + self.inner_query("", 0, params).map(|(buf, _)| { + Rows::new(self, buf.into_iter().collect()) + }) + } + + /// Executes the prepared statement, returning a lazily loaded iterator + /// over the resulting rows. + /// + /// No more than `row_limit` rows will be stored in memory at a time. Rows + /// will be pulled from the database in batches of `row_limit` as needed. + /// If `row_limit` is less than or equal to 0, `lazy_query` is equivalent + /// to `query`. + /// + /// This can only be called inside of a transaction, and the `Transaction` + /// object representing the active transaction must be passed to + /// `lazy_query`. + /// + /// ## Panics + /// + /// Panics if the provided `Transaction` is not associated with the same + /// `Connection` as this `Statement`, if the `Transaction` is not + /// active, or if the number of parameters provided does not match the + /// number of parameters expected. + pub fn lazy_query<'trans, 'stmt>(&'stmt self, + trans: &'trans Transaction, + params: &[&ToSql], + row_limit: i32) + -> Result> { + assert!(self.conn as *const _ == trans.conn as *const _, + "the `Transaction` passed to `lazy_query` must be associated with the same \ + `Connection` as the `Statement`"); + let conn = self.conn.conn.borrow(); + check_desync!(conn); + assert!(conn.trans_depth == trans.depth, + "`lazy_query` must be passed the active transaction"); + drop(conn); + + let id = self.next_portal_id.get(); + self.next_portal_id.set(id + 1); + let portal_name = format!("{}p{}", self.name, id); + + self.inner_query(&portal_name, row_limit, params).map(move |(data, more_rows)| { + LazyRows::new(self, data, portal_name, row_limit, more_rows, false, trans) + }) + } + + /// Executes a `COPY FROM STDIN` statement, returning the number of rows + /// added. + /// + /// The contents of the provided reader are passed to the Postgres server + /// verbatim; it is the caller's responsibility to ensure it uses the + /// proper format. See the + /// [Postgres documentation](http://www.postgresql.org/docs/9.4/static/sql-copy.html) + /// for details. + /// + /// If the statement is not a `COPY FROM STDIN` statement it will still be + /// executed and this method will return an error. + /// + /// # Examples + /// + /// ```rust,no_run + /// # use postgres::{Connection, SslMode}; + /// # let conn = Connection::connect("", &SslMode::None).unwrap(); + /// conn.batch_execute("CREATE TABLE people (id INT PRIMARY KEY, name VARCHAR)").unwrap(); + /// let stmt = conn.prepare("COPY people FROM STDIN").unwrap(); + /// stmt.copy_in(&[], &mut "1\tjohn\n2\tjane\n".as_bytes()).unwrap(); + /// ``` + pub fn copy_in(&self, params: &[&ToSql], r: &mut R) -> Result { + try!(self.inner_execute("", 0, params)); + let mut conn = self.conn.conn.borrow_mut(); + + match try!(conn.read_message()) { + CopyInResponse { .. } => {} + _ => { + loop { + match try!(conn.read_message()) { + ReadyForQuery { .. } => { + return Err(Error::IoError(io::Error::new( + io::ErrorKind::InvalidInput, + "called `copy_in` on a non-`COPY FROM STDIN` statement"))); + } + _ => {} + } + } + } + } + + let mut buf = [0; 16 * 1024]; + loop { + match fill_copy_buf(&mut buf, r, &SessionInfo::new(&conn)) { + Ok(0) => break, + Ok(len) => { + try_desync!(conn, conn.stream.write_message( + &CopyData { + data: &buf[..len], + })); + } + Err(err) => { + try!(conn.write_messages(&[ + CopyFail { + message: "", + }, + CopyDone, + Sync])); + match try!(conn.read_message()) { + ErrorResponse { .. } => { /* expected from the CopyFail */ } + _ => { + conn.desynchronized = true; + return Err(Error::IoError(bad_response())); + } + } + try!(conn.wait_for_ready()); + return Err(Error::IoError(err)); + } + } + } + + try!(conn.write_messages(&[CopyDone, Sync])); + + let num = match try!(conn.read_message()) { + CommandComplete { tag } => util::parse_update_count(tag), + ErrorResponse { fields } => { + try!(conn.wait_for_ready()); + return DbError::new(fields); + } + _ => { + conn.desynchronized = true; + return Err(Error::IoError(bad_response())); + } + }; + + try!(conn.wait_for_ready()); + Ok(num) + } + + /// Consumes the statement, clearing it from the Postgres session. + /// + /// If this statement was created via the `prepare_cached` method, `finish` + /// does nothing. + /// + /// Functionally identical to the `Drop` implementation of the + /// `Statement` except that it returns any error to the caller. + pub fn finish(mut self) -> Result<()> { + self.finish_inner() + } +} + +fn fill_copy_buf(buf: &mut [u8], r: &mut R, info: &SessionInfo) + -> io::Result { + let mut nread = 0; + while nread < buf.len() { + match r.read_with_info(&mut buf[nread..], info) { + Ok(0) => break, + Ok(n) => nread += n, + Err(ref e) if e.kind() == io::ErrorKind::Interrupted => {} + Err(e) => return Err(e), + } + } + Ok(nread) +} + +/// Information about a column of the result of a query. +#[derive(PartialEq, Eq, Clone, Debug)] +pub struct Column { + name: String, + type_: Type +} + +impl ColumnNew for Column { + fn new(name: String, type_: Type) -> Column { + Column { + name: name, + type_: type_, + } + } +} + +impl Column { + /// The name of the column. + pub fn name(&self) -> &str { + &self.name + } + + /// The type of the data in the column. + pub fn type_(&self) -> &Type { + &self.type_ + } +} + +