diff -Ncr -X xCVS clean/org/postgresql/core/PGStream.java restartStream/org/postgresql/core/PGStream.java
*** clean/org/postgresql/core/PGStream.java	Mon Sep 20 01:36:48 2004
--- restartStream/org/postgresql/core/PGStream.java	Fri Oct 15 12:35:20 2004
***************
*** 24,29 ****
--- 24,31 ----
  import java.net.Socket;
  import java.sql.*;
  
+ import org.postgresql.util.GT;
+ 
  /**
   * Wrapper around the raw connection to the server that implements some basic
   * primitives (reading/writing formatted data, doing string encoding, etc).
***************
*** 437,442 ****
--- 439,445 ----
  	 * @param remaining the number of bytes to copy
  	 */
  	public void SendStream(InputStream inStream, int remaining) throws IOException {
+ 		int expectedLength = remaining;
  		if (streamBuffer == null)
  			streamBuffer = new byte[8192];
  
***************
*** 444,453 ****
  			int count = (remaining > streamBuffer.length ? streamBuffer.length : remaining);
  			int readCount;
  
! 			readCount = inStream.read(streamBuffer, 0, count);
! 			if (readCount < 0)
! 				throw new EOFException("Premature end of input stream");
! 			
  			Send(streamBuffer, readCount);
  			remaining -= readCount;
  		}
--- 447,465 ----
  			int count = (remaining > streamBuffer.length ? streamBuffer.length : remaining);
  			int readCount;
  
! 			try {
! 				readCount = inStream.read(streamBuffer, 0, count);
! 				if (readCount < 0)
! 					throw new EOFException(GT.tr("Premature end of input stream, expected {0} bytes, but only read {1}.", new Object[]{new Integer(expectedLength), new Integer(expectedLength - remaining)}));
! 			} catch(IOException ioe) {
! 				while (remaining > 0) {
! 					Send(streamBuffer, count);
! 					remaining -= count;
! 					count = (remaining > streamBuffer.length ? streamBuffer.length : remaining);
! 				}
! 				throw new PGStreamingException(ioe);
! 			}
! 
  			Send(streamBuffer, readCount);
  			remaining -= readCount;
  		}
diff -Ncr -X xCVS clean/org/postgresql/core/PGStreamingException.java restartStream/org/postgresql/core/PGStreamingException.java
*** clean/org/postgresql/core/PGStreamingException.java	Wed Dec 31 16:00:00 1969
--- restartStream/org/postgresql/core/PGStreamingException.java	Fri Oct 15 11:22:27 2004
***************
*** 0 ****
--- 1,16 ----
+ package org.postgresql.core;
+ 
+ import java.io.IOException;
+ 
+ public class PGStreamingException extends IOException {
+ 
+ 	private IOException _ioe;
+ 
+ 	public PGStreamingException(IOException ioe) {
+ 		_ioe = ioe;
+ 	}
+ 
+ 	public IOException getIOException() {
+ 		return _ioe;
+ 	}
+ }
diff -Ncr -X xCVS clean/org/postgresql/core/v3/QueryExecutorImpl.java restartStream/org/postgresql/core/v3/QueryExecutorImpl.java
*** clean/org/postgresql/core/v3/QueryExecutorImpl.java	Sun Oct 10 08:39:37 2004
--- restartStream/org/postgresql/core/v3/QueryExecutorImpl.java	Fri Oct 15 12:45:46 2004
***************
*** 131,140 ****
  		((V3ParameterList)parameters).checkAllParametersSet();
  
  		try {
! 			handler = sendQueryPreamble(handler, flags);		
! 			sendQuery((V3Query)query, (V3ParameterList)parameters, maxRows, fetchSize, flags);		
! 			sendSync();
! 			processResults(handler, flags);
  		} catch (IOException e) {
  			protoConnection.close();
  			handler.handleError(new PSQLException(GT.tr("An I/O error occured while sending to the backend."), PSQLState.CONNECTION_FAILURE, e));
--- 131,153 ----
  		((V3ParameterList)parameters).checkAllParametersSet();
  
  		try {
! 			try {
! 				handler = sendQueryPreamble(handler, flags);		
! 				sendQuery((V3Query)query, (V3ParameterList)parameters, maxRows, fetchSize, flags);		
! 				sendSync();
! 				processResults(handler, flags);
! 			} catch (PGStreamingException se) {
! 				// If a streaming error has happened it will
! 				// send the Bind, but not the Execute message
! 				// so we can just continue on as if nothing
! 				// has happened.  Perhaps we need to
! 				// introduce an error here to force the
! 				// caller to rollback if there is a
! 				// transaction in progress?
! 				sendSync();
! 				processResults(handler, flags);
! 				handler.handleError(new PSQLException(GT.tr("An I/O error occured on the provided InputStream."), PSQLState.INVALID_PARAMETER_VALUE, se.getIOException()));
! 			}
  		} catch (IOException e) {
  			protoConnection.close();
  			handler.handleError(new PSQLException(GT.tr("An I/O error occured while sending to the backend."), PSQLState.CONNECTION_FAILURE, e));
***************
*** 605,616 ****
  			pgStream.SendInteger2(params.isBinary(i) ? 1 : 0);  // Parameter format code
  			
  		pgStream.SendInteger2(params.getParameterCount());      // # of parameter values
  		for (int i = 1; i <= params.getParameterCount(); ++i) {
  			if (params.isNull(i))
  				pgStream.SendInteger4(-1);                      // Magic size of -1 means NULL
  			else {
  				pgStream.SendInteger4(params.getV3Length(i));   // Parameter size
! 				params.writeV3Value(i, pgStream);                 // Parameter value
  			}
  		}
  
--- 618,643 ----
  			pgStream.SendInteger2(params.isBinary(i) ? 1 : 0);  // Parameter format code
  			
  		pgStream.SendInteger2(params.getParameterCount());      // # of parameter values
+ 
+ 		// If an error occurs when reading a stream we have to
+ 		// continue pumping out data to match the length we
+ 		// said we would.  Once we've done that we throw
+ 		// this exception.  Multiple exceptions can occur and
+ 		// it really doesn't matter which one is reported back
+ 		// to the caller.
+ 		//
+ 		PGStreamingException streamException = null;
+ 		
  		for (int i = 1; i <= params.getParameterCount(); ++i) {
  			if (params.isNull(i))
  				pgStream.SendInteger4(-1);                      // Magic size of -1 means NULL
  			else {
  				pgStream.SendInteger4(params.getV3Length(i));   // Parameter size
! 				try {
! 					params.writeV3Value(i, pgStream);                 // Parameter value
! 				} catch(PGStreamingException se) {
! 					streamException = se;
! 				}
  			}
  		}
  
***************
*** 618,623 ****
--- 645,654 ----
  		pgStream.SendChar(0);       //  (...)
  
  		pendingBindQueue.add(portal);
+ 
+ 		if (streamException != null) {
+ 			throw streamException;
+ 		}
  	}
  
  	private void sendDescribe(Portal portal) throws IOException {
diff -Ncr -X xCVS clean/org/postgresql/test/jdbc2/PreparedStatementTest.java restartStream/org/postgresql/test/jdbc2/PreparedStatementTest.java
*** clean/org/postgresql/test/jdbc2/PreparedStatementTest.java	Mon Jun 28 23:43:28 2004
--- restartStream/org/postgresql/test/jdbc2/PreparedStatementTest.java	Fri Oct 15 12:12:00 2004
***************
*** 1,6 ****
--- 1,7 ----
  package org.postgresql.test.jdbc2;
  
  import org.postgresql.test.TestUtil;
+ import org.postgresql.test.util.BrokenInputStream;
  import junit.framework.TestCase;
  import java.io.*;
  import java.sql.*;
***************
*** 87,92 ****
--- 88,141 ----
  		} catch (SQLException e) {}
  	}
  
+ 	public void testBinaryStreamErrorsRestartable() throws SQLException {
+ 		// The V2 protocol does not have the ability to recover when
+ 		// streaming data to the server.  We could potentially try
+ 		// introducing a syntax error to force the query to fail, but
+ 		// that seems dangerous.
+ 		//
+ 		if(!TestUtil.haveMinimumServerVersion(conn, "7.4")) {
+ 			return;
+ 		}
+ 
+ 		byte buf[] = new byte[10];
+ 		for (int i=0; i<buf.length; i++) {
+ 			buf[i] = (byte)i;
+ 		}
+ 
+ 		// Send a shorter InputStream than the length argument implies.
+ 		InputStream is = new ByteArrayInputStream(buf);
+ 		runBrokenStream(is, buf.length+1);
+ 
+ 		// Send an InputStream that is designed to throw an Exception
+ 		// during read.
+ 		is = new BrokenInputStream(new ByteArrayInputStream(buf), buf.length/2);
+ 		runBrokenStream(is, buf.length);
+ 	}
+ 
+ 	private void runBrokenStream(InputStream is, int length) throws SQLException
+ 	{
+ 		PreparedStatement pstmt = null;
+ 		try {
+ 			pstmt = conn.prepareStatement("INSERT INTO streamtable (bin,str) VALUES (?,?)");
+ 			pstmt.setBinaryStream(1, is, length);
+ 			pstmt.setString(2, "Other");
+ 			pstmt.executeUpdate();
+ 			fail("This isn't supposed to work.");
+ 		} catch (SQLException sqle) {
+ 			// don't need to rollback because we're in autocommit mode
+ 			pstmt.close();
+ 
+ 			// verify the connection is still valid and the row didn't go in.
+ 			Statement stmt = conn.createStatement();
+ 			ResultSet rs = stmt.executeQuery("SELECT COUNT(*) FROM streamtable");
+ 			assertTrue(rs.next());
+ 			assertEquals(0, rs.getInt(1));
+ 			rs.close();
+ 			stmt.close();
+ 		}
+ 	}
+ 
  	private void doSetBinaryStream(ByteArrayInputStream bais, int length) throws SQLException
  	{
  		PreparedStatement pstmt = conn.prepareStatement("INSERT INTO streamtable (bin,str) VALUES (?,?)");
diff -Ncr -X xCVS clean/org/postgresql/test/util/BrokenInputStream.java restartStream/org/postgresql/test/util/BrokenInputStream.java
*** clean/org/postgresql/test/util/BrokenInputStream.java	Wed Dec 31 16:00:00 1969
--- restartStream/org/postgresql/test/util/BrokenInputStream.java	Fri Oct 15 10:49:57 2004
***************
*** 0 ****
--- 1,25 ----
+ package org.postgresql.test.util;
+ 
+ import java.io.InputStream;
+ import java.io.IOException;
+ 
+ public class BrokenInputStream extends InputStream {
+ 
+ 	private InputStream _is;
+ 	private long _numRead;
+ 	private long _breakOn;
+ 
+ 	public BrokenInputStream(InputStream is, long breakOn) {
+ 		_is = is;
+ 		_breakOn = breakOn;
+ 		_numRead = 0;
+ 	}
+ 
+ 	public int read() throws IOException {
+ 		if (_breakOn > _numRead++) {
+ 			throw new IOException("I was told to break on " + _breakOn);
+ 		}
+ 
+ 		return _is.read();
+ 	}
+ }