dnim/rust-postgres
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'release-v0.9.6' into release

Browse Source
This commit is contained in:
Steven Fackler 2015-08-28 17:02:12 -07:00
commit fc2ae1da54
11 changed files with 782 additions and 422 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
Cargo.toml
View File

@ -1,11 +1,11 @@
[package]
name = "postgres"
version = "0.9.5"
version = "0.9.6"
authors = ["Steven Fackler <sfackler@gmail.com>"]
license = "MIT"
description = "A native PostgreSQL driver"
repository = "https://github.com/sfackler/rust-postgres"
documentation = "https://sfackler.github.io/rust-postgres/doc/v0.9.5/postgres"
documentation = "https://sfackler.github.io/rust-postgres/doc/v0.9.6/postgres"
readme = "README.md"
keywords = ["database", "sql"]
build = "build.rs"
@ -31,7 +31,7 @@ log = "0.3"
phf = "0.7"
rustc-serialize = "0.3"
chrono = { version = "0.2.14", optional = true }
openssl = { version = "0.6", optional = true }
openssl = { version = "0.6.4", optional = true }
serde = { version = "0.3", optional = true }
time = { version = "0.1.14", optional = true }
unix_socket = { version = ">= 0.3, < 0.5", optional = true }

2
README.md
View File

@ -1,7 +1,7 @@
# Rust-Postgres
A native PostgreSQL driver for Rust.
[Documentation](https://sfackler.github.io/rust-postgres/doc/v0.9.5/postgres)
[Documentation](https://sfackler.github.io/rust-postgres/doc/v0.9.6/postgres)
[![Build Status](https://travis-ci.org/sfackler/rust-postgres.png?branch=master)](https://travis-ci.org/sfackler/rust-postgres) [![Latest Version](https://img.shields.io/crates/v/postgres.svg)](https://crates.io/crates/postgres)

6
src/error.rs
View File

@ -350,3 +350,9 @@ impl From<byteorder::Error> for Error {
Error::IoError(From::from(err))
}
}
impl From<Error> for io::Error {
fn from(err: Error) -> io::Error {
io::Error::new(io::ErrorKind::Other, err)
}
}

2
src/io/openssl.rs
View File

@ -18,7 +18,7 @@ impl StreamWrapper for SslStream<Stream> {
impl NegotiateSsl for SslContext {
fn negotiate_ssl(&self, _: &str, stream: Stream)
-> Result<Box<StreamWrapper>, Box<Error+Send+Sync>> {
let stream = try!(SslStream::new(self, stream));
let stream = try!(SslStream::connect(self, stream));
Ok(Box::new(stream))
}
}

439
src/lib.rs
View File

@ -41,7 +41,7 @@
//! }
//! }
//! ```
#![doc(html_root_url="https://sfackler.github.io/rust-postgres/doc/v0.9.5")]
#![doc(html_root_url="https://sfackler.github.io/rust-postgres/doc/v0.9.6")]
#![warn(missing_docs)]
extern crate bufstream;
@ -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 url;
mod util;
pub mod types;
pub mod error;
pub mod io;
pub mod rows;
mod md5;
pub mod stmt;
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<Statement<'a>> {
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<Statement<'a>> {
@ -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<u64> {
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,372 +1279,6 @@ impl<'conn> Transaction<'conn> {
}
}
/// A prepared statement.
pub struct Statement<'conn> {
conn: &'conn Connection,
name: String,
param_types: Vec<Type>,
columns: Vec<Column>,
next_portal_id: Cell<u32>,
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<Vec<Option<Vec<u8>>>>, 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<u64> {
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]));
}
_ => {
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<Rows<'a>> {
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<LazyRows<'trans, 'stmt>> {
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<R: ReadWithInfo>(&self, params: &[&ToSql], r: &mut R) -> Result<u64> {
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<R: ReadWithInfo>(buf: &mut [u8], r: &mut R, info: &SessionInfo)
-> std_io::Result<usize> {
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<Vec<Option<Vec<u8>>>>) -> Result<bool> {
let more_rows;
loop {
@ -1690,6 +1303,17 @@ fn read_rows(conn: &mut InnerConnection, buf: &mut VecDeque<Vec<Option<Vec<u8>>>
},
Sync]));
}
CopyOutResponse { .. } => {
loop {
match try!(conn.read_message()) {
ReadyForQuery { .. } => break,
_ => {}
}
}
return Err(Error::IoError(std_io::Error::new(
std_io::ErrorKind::InvalidInput,
"COPY queries cannot be directly executed")));
}
_ => {
conn.desynchronized = true;
return Err(Error::IoError(bad_response()));
@ -1804,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<Type>,
columns: Vec<Column>,
next_portal_id: Cell<u32>,
finished: bool) -> Statement<'conn>;
fn conn(&self) -> &'conn Connection;
}
trait ColumnNew {
fn new(name: String, type_: Type) -> Column;
}

36
src/message.rs
View File

@ -32,10 +32,19 @@ pub enum BackendMessage {
CommandComplete {
tag: String,
},
// FIXME naming
BCopyData {
data: Vec<u8>,
},
BCopyDone,
CopyInResponse {
format: u8,
column_formats: Vec<u16>,
},
CopyOutResponse {
format: u8,
column_formats: Vec<u16>,
},
DataRow {
row: Vec<Option<Vec<u8>>>
},
@ -292,7 +301,15 @@ impl<R: BufRead> ReadMessage for R {
channel: try!(rdr.read_cstr()),
payload: try!(rdr.read_cstr())
},
b'c' => BCopyDone,
b'C' => CommandComplete { tag: try!(rdr.read_cstr()) },
b'd' => {
let mut data = vec![];
try!(rdr.read_to_end(&mut data));
BCopyData {
data: data,
}
}
b'D' => try!(read_data_row(&mut rdr)),
b'E' => ErrorResponse { fields: try!(read_fields(&mut rdr)) },
b'G' => {
@ -306,6 +323,17 @@ impl<R: BufRead> ReadMessage for R {
column_formats: column_formats,
}
}
b'H' => {
let format = try!(rdr.read_u8());
let mut column_formats = vec![];
for _ in 0..try!(rdr.read_u16::<BigEndian>()) {
column_formats.push(try!(rdr.read_u16::<BigEndian>()));
}
CopyOutResponse {
format: format,
column_formats: column_formats,
}
}
b'I' => EmptyQueryResponse,
b'K' => BackendKeyData {
process_id: try!(rdr.read_u32::<BigEndian>()),
@ -322,7 +350,8 @@ impl<R: BufRead> ReadMessage for R {
b't' => try!(read_parameter_description(&mut rdr)),
b'T' => try!(read_row_description(&mut rdr)),
b'Z' => ReadyForQuery { _state: try!(rdr.read_u8()) },
_ => return Err(io::Error::new(io::ErrorKind::Other, "unexpected message tag")),
t => return Err(io::Error::new(io::ErrorKind::Other,
format!("unexpected message tag `{}`", t))),
};
if rdr.limit() != 0 {
return Err(io::Error::new(io::ErrorKind::Other, "didn't read entire message"));
@ -377,7 +406,8 @@ fn read_auth_message<R: Read>(buf: &mut R) -> io::Result<BackendMessage> {
6 => AuthenticationSCMCredential,
7 => AuthenticationGSS,
9 => AuthenticationSSPI,
_ => return Err(io::Error::new(io::ErrorKind::Other, "unexpected authentication tag")),
t => return Err(io::Error::new(io::ErrorKind::Other,
format!("unexpected authentication tag `{}`", t))),
})
}
@ -411,7 +441,7 @@ fn read_row_description<R: BufRead>(buf: &mut R) -> io::Result<BackendMessage> {
Ok(RowDescription { descriptions: types })
}
trait FromUsize {
trait FromUsize: Sized {
fn from_usize(x: usize) -> io::Result<Self>;
}

17
src/rows.rs
View File

@ -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<I, T>(&self, idx: I) -> Result<T> 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 !<T as FromSql>::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<usize> {
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<usize> {
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 {

628
src/stmt.rs Normal file
View File

@ -0,0 +1,628 @@
//! Prepared statements
use debug_builders::DebugStruct;
use std::cell::{Cell, RefMut};
use std::collections::VecDeque;
use std::fmt;
use std::io::{self, Cursor, BufRead, Read};
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 {InnerConnection, SessionInfoNew, LazyRowsNew, DbErrorNew, ColumnNew};
/// A prepared statement.
pub struct Statement<'conn> {
conn: &'conn Connection,
name: String,
param_types: Vec<Type>,
columns: Vec<Column>,
next_portal_id: Cell<u32>,
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<Type>,
columns: Vec<Column>,
next_portal_id: Cell<u32>,
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<Vec<Option<Vec<u8>>>>, 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<u64> {
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<Rows<'a>> {
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<LazyRows<'trans, 'stmt>> {
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<R: ReadWithInfo>(&self, params: &[&ToSql], r: &mut R) -> Result<u64> {
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)
}
/// Executes a `COPY TO STDOUT` statement, returning a `Read`er of the
/// resulting data.
///
/// See the [Postgres documentation](http://www.postgresql.org/docs/9.4/static/sql-copy.html)
/// for details on the data format.
///
/// If the statement is not a `COPY TO STDOUT` statement it will still be
/// executed and this method will return an error.
///
/// # Warning
///
/// The underlying connection may not be used while the returned `Read`er
/// exists. Any attempt to do so will panic.
///
/// # Examples
///
/// ```rust,no_run
/// # use std::io::Read;
/// # use postgres::{Connection, SslMode};
/// # let conn = Connection::connect("", &SslMode::None).unwrap();
/// conn.batch_execute("
/// CREATE TABLE people (id INT PRIMARY KEY, name VARCHAR);
/// INSERT INTO people (id, name) VALUES (1, 'john'), (2, 'jane');").unwrap();
/// let stmt = conn.prepare("COPY people TO STDOUT").unwrap();
/// let mut r = stmt.copy_out(&[]).unwrap();
/// let mut buf = vec![];
/// r.read_to_end(&mut buf).unwrap();
/// r.finish().unwrap();
/// assert_eq!(buf, b"1\tjohn\n2\tjane\n");
/// ```
pub fn copy_out<'a>(&'a self, params: &[&ToSql]) -> Result<CopyOutReader<'a>> {
try!(self.inner_execute("", 0, params));
let mut conn = self.conn.conn.borrow_mut();
let (format, column_formats) = match try!(conn.read_message()) {
CopyOutResponse { format, column_formats } => (format, column_formats),
CopyInResponse { .. } => {
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(io::Error::new(
io::ErrorKind::InvalidInput,
"called `copy_out` on a non-`COPY TO STDOUT` statement")));
}
_ => {
loop {
match try!(conn.read_message()) {
ReadyForQuery { .. } => {
return Err(Error::IoError(io::Error::new(
io::ErrorKind::InvalidInput,
"called `copy_out` on a non-`COPY TO STDOUT` statement")));
}
_ => {}
}
}
}
};
Ok(CopyOutReader {
conn: conn,
format: Format::from_u16(format as u16),
column_formats: column_formats.iter().map(|&f| Format::from_u16(f)).collect(),
buf: Cursor::new(vec![]),
finished: false,
})
}
/// 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<R: ReadWithInfo>(buf: &mut [u8], r: &mut R, info: &SessionInfo)
-> io::Result<usize> {
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_
}
}
/// The format of a portion of COPY query data.
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
pub enum Format {
/// A text based format.
Text,
/// A binary format.
Binary,
}
impl Format {
fn from_u16(value: u16) -> Format {
match value {
0 => Format::Text,
_ => Format::Binary,
}
}
}
/// A `Read`er for data from `COPY TO STDOUT` queries.
///
/// # Warning
///
/// The underlying connection may not be used while a `CopyOutReader` exists.
/// Any attempt to do so will panic.
pub struct CopyOutReader<'a> {
conn: RefMut<'a, InnerConnection>,
format: Format,
column_formats: Vec<Format>,
buf: Cursor<Vec<u8>>,
finished: bool,
}
impl<'a> Drop for CopyOutReader<'a> {
fn drop(&mut self) {
let _ = self.finish_inner();
}
}
impl<'a> CopyOutReader<'a> {
/// Returns the format of the overall data.
pub fn format(&self) -> Format {
self.format
}
/// Returns the format of the individual columns.
pub fn column_formats(&self) -> &[Format] {
&self.column_formats
}
/// Returns session info for the associated connection.
pub fn session_info<'b>(&'b self) -> SessionInfo<'b> {
SessionInfo::new(&*self.conn)
}
/// Consumes the `CopyOutReader`, throwing away any unread data.
///
/// Functionally equivalent to `CopyOutReader`'s `Drop` implementation,
/// except that it returns any error encountered to the caller.
pub fn finish(mut self) -> Result<()> {
self.finish_inner()
}
fn finish_inner(&mut self) -> Result<()> {
while !self.finished {
let pos = self.buf.get_ref().len() as u64;
self.buf.set_position(pos);
try!(self.ensure_filled());
}
Ok(())
}
fn ensure_filled(&mut self) -> Result<()> {
if self.finished || self.buf.position() != self.buf.get_ref().len() as u64 {
return Ok(());
}
match try!(self.conn.read_message()) {
BCopyData { data } => self.buf = Cursor::new(data),
BCopyDone => {
self.finished = true;
match try!(self.conn.read_message()) {
CommandComplete { .. } => {}
_ => {
self.conn.desynchronized = true;
return Err(Error::IoError(bad_response()));
}
}
try!(self.conn.wait_for_ready());
}
ErrorResponse { fields } => {
self.finished = true;
try!(self.conn.wait_for_ready());
return DbError::new(fields);
}
_ => {
self.conn.desynchronized = true;
return Err(Error::IoError(bad_response()));
}
}
Ok(())
}
}
impl<'a> Read for CopyOutReader<'a> {
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
try!(self.ensure_filled());
self.buf.read(buf)
}
}
impl<'a> BufRead for CopyOutReader<'a> {
fn fill_buf(&mut self) -> io::Result<&[u8]> {
try!(self.ensure_filled());
self.buf.fill_buf()
}
fn consume(&mut self, amt: usize) {
self.buf.consume(amt)
}
}

19
src/types/chrono.rs
View File

@ -24,8 +24,14 @@ impl FromSql for NaiveDateTime {
impl ToSql for NaiveDateTime {
fn to_sql<W: Write+?Sized>(&self, _: &Type, mut w: &mut W, _: &SessionInfo) -> Result<IsNull> {
let t = (*self - base()).num_microseconds().unwrap();
try!(w.write_i64::<BigEndian>(t));
let time = match (*self - base()).num_microseconds() {
Some(time) => time,
None => {
let err: Box<error::Error+Sync+Send> = "value too large to transmit".into();
return Err(Error::Conversion(err));
}
};
try!(w.write_i64::<BigEndian>(time));
Ok(IsNull::No)
}
@ -128,7 +134,14 @@ impl FromSql for NaiveTime {
impl ToSql for NaiveTime {
fn to_sql<W: Write+?Sized>(&self, _: &Type, mut w: &mut W, _: &SessionInfo) -> Result<IsNull> {
let delta = *self - NaiveTime::from_hms(0, 0, 0);
try!(w.write_i64::<BigEndian>(delta.num_microseconds().unwrap()));
let time = match delta.num_microseconds() {
Some(time) => time,
None => {
let err: Box<error::Error+Sync+Send> = "value too large to transmit".into();
return Err(Error::Conversion(err));
}
};
try!(w.write_i64::<BigEndian>(time));
Ok(IsNull::No)
}

5
src/types/mod.rs
View File

@ -142,7 +142,6 @@ macro_rules! make_postgres_type {
_ => None
}
}
}
impl Type {
@ -584,9 +583,9 @@ impl error::Error for WasNull {
/// `Option<T>` where `T` implements `FromSql`. An `Option<T>` represents a
/// nullable Postgres value.
pub trait FromSql: Sized {
/// Creates a new value of this type from a `Read` of Postgres data.
/// Creates a new value of this type from a `Read`er of Postgres data.
///
/// If the value was `NULL`, the `Read` will be `None`.
/// If the value was `NULL`, the `Read`er will be `None`.
///
/// The caller of this method is responsible for ensuring that this type
/// is compatible with the Postgres `Type`.

44
tests/test.rs
View File

@ -8,6 +8,7 @@ extern crate openssl;
use openssl::ssl::{SslContext, SslMethod};
use std::thread;
use std::io;
use std::io::prelude::*;
use postgres::{HandleNotice,
Notification,
@ -756,6 +757,49 @@ fn test_copy() {
stmt.query(&[]).unwrap().iter().map(|r| r.get(0)).collect::<Vec<i32>>());
}
#[test]
fn test_query_copy_out_err() {
let conn = or_panic!(Connection::connect("postgres://postgres@localhost", &SslMode::None));
or_panic!(conn.batch_execute("
CREATE TEMPORARY TABLE foo (id INT);
INSERT INTO foo (id) VALUES (0), (1), (2), (3)"));
let stmt = or_panic!(conn.prepare("COPY foo (id) TO STDOUT"));
match stmt.query(&[]) {
Ok(_) => panic!("unexpected success"),
Err(Error::IoError(ref e)) if e.to_string().contains("COPY") => {}
Err(e) => panic!("unexpected error {:?}", e),
}
}
#[test]
fn test_copy_out() {
let conn = or_panic!(Connection::connect("postgres://postgres@localhost", &SslMode::None));
or_panic!(conn.batch_execute("
CREATE TEMPORARY TABLE foo (id INT);
INSERT INTO foo (id) VALUES (0), (1), (2), (3)"));
let stmt = or_panic!(conn.prepare("COPY (SELECT id FROM foo ORDER BY id) TO STDOUT"));
let mut reader = or_panic!(stmt.copy_out(&[]));
let mut out = vec![];
or_panic!(reader.read_to_end(&mut out));
assert_eq!(out, b"0\n1\n2\n3\n");
drop(reader);
or_panic!(conn.batch_execute("SELECT 1"));
}
#[test]
fn test_copy_out_partial_read() {
let conn = or_panic!(Connection::connect("postgres://postgres@localhost", &SslMode::None));
or_panic!(conn.batch_execute("
CREATE TEMPORARY TABLE foo (id INT);
INSERT INTO foo (id) VALUES (0), (1), (2), (3)"));
let stmt = or_panic!(conn.prepare("COPY (SELECT id FROM foo ORDER BY id) TO STDOUT"));
let mut reader = or_panic!(stmt.copy_out(&[]));
let mut out = vec![];
or_panic!(reader.by_ref().take(5).read_to_end(&mut out));
assert_eq!(out, b"0\n1\n2");
drop(reader);
or_panic!(conn.batch_execute("SELECT 1"));
}
#[test]
// Just make sure the impls don't infinite loop