Fix for Bug#17881458, BEHAVIOR OF SETBINARYSTREAM() METHOD IS DIFFERENT WHEN USESERVERPREPSTMTS=TRUE.

Change-Id: Ic33f8fc7dfd43253f9091f128a4c57e494dea86e

Author: fjssilva

CHANGES

@@ -3,6 +3,8 @@
 
 Version 9.5.0
 
+  - Fix for Bug#17881458, BEHAVIOR OF SETBINARYSTREAM() METHOD IS DIFFERENT WHEN USESERVERPREPSTMTS=TRUE.
+
   - Fix for Bug#45554 (Bug#11754018), Connector/J does not encode binary data if useServerPrepStatements=false.
 
   - Fix for Bug#114974 (Bug#36614381), the SQL in batch will not clear after statement close.

src/main/core-api/java/com/mysql/cj/util/Util.java

@@ -432,8 +432,7 @@ public static boolean isRunningOnWindows() {
     }
 
     /**
-     * Reads length bytes from reader into buf. Blocks until enough input is
-     * available
+     * Reads length bytes from reader into buf. Blocks until enough input is available.
      *
      * @param reader
      *            {@link Reader}
@@ -459,24 +458,90 @@ public static int readFully(Reader reader, char[] buf, int length) throws IOExce
         return numCharsRead;
     }
 
-    public static final int readBlock(InputStream i, byte[] b, ExceptionInterceptor exceptionInterceptor) {
+    /**
+     * Reads data from the provided {@link InputStream} into the specified buffer.
+     *
+     * @param inStream
+     *            The {@link InputStream} to read from.
+     * @param buffer
+     *            Buffer into which bytes are read.
+     * @param exceptionInterceptor
+     *            The {@link ExceptionInterceptor} to handle new exceptions thrown.
+     * @return
+     *         The number of bytes read, or -1 if the end of the stream is reached
+     */
+    public static final int readBlock(InputStream inStream, byte[] buffer, ExceptionInterceptor exceptionInterceptor) {
         try {
-            return i.read(b);
-        } catch (Throwable ex) {
+            return inStream.read(buffer);
+        } catch (Exception ex) {
             throw ExceptionFactory.createException(Messages.getString("Util.5") + ex.getClass().getName(), exceptionInterceptor);
         }
     }
 
-    public static final int readBlock(InputStream i, byte[] b, int length, ExceptionInterceptor exceptionInterceptor) {
+    /**
+     * Reads data from the provided {@link InputStream} into the specified buffer, up to the specified length or the size of the buffer, whichever is smaller.
+     *
+     * @param inStream
+     *            The {@link InputStream} to read from.
+     * @param buffer
+     *            Buffer into which bytes are read.
+     * @param length
+     *            The maximum amount of bytes to read from the input stream.
+     * @param exceptionInterceptor
+     *            The {@link ExceptionInterceptor} to handle new exceptions thrown.
+     * @return
+     *         The number of bytes read, or -1 if the end of the stream is reached
+     */
+    public static final int readBlock(InputStream inStream, byte[] buffer, long length, ExceptionInterceptor exceptionInterceptor) {
         try {
-            int lengthToRead = length;
-            if (lengthToRead > b.length) {
-                lengthToRead = b.length;
-            }
-            return i.read(b, 0, lengthToRead);
-        } catch (Throwable ex) {
+            int lengthToRead = length > buffer.length ? buffer.length : (int) length;
+            return inStream.read(buffer, 0, lengthToRead);
+        } catch (Exception ex) {
             throw ExceptionFactory.createException(Messages.getString("Util.5") + ex.getClass().getName(), exceptionInterceptor);
         }
     }
 
+    /**
+     * Reads characters from the provided {@link Reader} into the specified buffer.
+     *
+     * @param reader
+     *            The {@link Reader} to read from.
+     * @param buffer
+     *            Buffer into which the characters are read.
+     * @param exceptionInterceptor
+     *            The {@link ExceptionInterceptor} to handle new exceptions thrown.
+     * @return
+     *         The number of characters read, or -1 if the end of the stream is reached
+     */
+    public static final int readBlock(Reader reader, char[] buffer, ExceptionInterceptor exceptionInterceptor) {
+        try {
+            return reader.read(buffer);
+        } catch (Exception e) {
+            throw ExceptionFactory.createException(Messages.getString("Util.6") + e.getClass().getName(), exceptionInterceptor);
+        }
+    }
+
+    /**
+     * Reads characters from the provided {@link Reader} into the specified buffer, up to the specified length or the size of the buffer, whichever is smaller.
+     *
+     * @param reader
+     *            The {@link Reader} to read from.
+     * @param buffer
+     *            Buffer into which the characters are read.
+     * @param length
+     *            The maximum amount of characters to read from the stream.
+     * @param exceptionInterceptor
+     *            The {@link ExceptionInterceptor} to handle new exceptions thrown.
+     * @return
+     *         The number of characters read, or -1 if the end of the stream is reached
+     */
+    public static final int readBlock(Reader reader, char[] buffer, long length, ExceptionInterceptor exceptionInterceptor) {
+        try {
+            int lengthToRead = length > buffer.length ? buffer.length : (int) length;
+            return reader.read(buffer, 0, lengthToRead);
+        } catch (Exception ex) {
+            throw ExceptionFactory.createException(Messages.getString("Util.6") + ex.getClass().getName(), exceptionInterceptor);
+        }
+    }
+
 }

src/main/core-impl/java/com/mysql/cj/ServerPreparedQuery.java

@@ -20,7 +20,6 @@
 
 package com.mysql.cj;
 
-import java.io.Closeable;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.Reader;
@@ -50,14 +49,15 @@
 import com.mysql.cj.telemetry.TelemetrySpan;
 import com.mysql.cj.telemetry.TelemetrySpanName;
 import com.mysql.cj.util.StringUtils;
+import com.mysql.cj.util.Util;
 
 // TODO should not be protocol-specific
 
 public class ServerPreparedQuery extends ClientPreparedQuery {
 
-    public static final int BLOB_STREAM_READ_BUF_SIZE = 8192;
-    public static final byte OPEN_CURSOR_FLAG = 0x01;
-    public static final byte PARAMETER_COUNT_AVAILABLE = 0x08;
+    private static final int READ_BUFFER_SIZE = 8192;
+    private static final byte OPEN_CURSOR_FLAG = 0x01;
+    private static final byte PARAMETER_COUNT_AVAILABLE = 0x08;
 
     /** The ID that the server uses to identify this PreparedStatement */
     private long serverStatementId;
@@ -409,24 +409,24 @@ private void serverLongData(int parameterIndex, BindValue binding) {
                     this.session.getProtocol()
                             .sendCommand(this.commandBuilder.buildComStmtSendLongData(packet, this.serverStatementId, parameterIndex, (byte[]) value), true, 0);
                 } else if (value instanceof InputStream) {
-                    storeStreamOrReader(parameterIndex, packet, (InputStream) value);
+                    storeStream(parameterIndex, packet, (InputStream) value, binding.getScaleOrLength());
                 } else if (value instanceof java.sql.Blob) {
                     try {
-                        storeStreamOrReader(parameterIndex, packet, ((java.sql.Blob) value).getBinaryStream());
+                        storeStream(parameterIndex, packet, ((java.sql.Blob) value).getBinaryStream(), binding.getScaleOrLength());
                     } catch (Throwable t) {
                         throw ExceptionFactory.createException(t.getMessage(), this.session.getExceptionInterceptor());
                     }
                 } else if (value instanceof Reader) {
                     if (binding.isNational() && !this.charEncoding.equalsIgnoreCase("UTF-8") && !this.charEncoding.equalsIgnoreCase("utf8")) {
                         throw ExceptionFactory.createException(Messages.getString("ServerPreparedStatement.31"), this.session.getExceptionInterceptor());
                     }
-                    storeStreamOrReader(parameterIndex, packet, (Reader) value);
+                    storeReader(parameterIndex, packet, (Reader) value, binding.getScaleOrLength());
                 } else if (value instanceof Clob) {
                     if (binding.isNational() && !this.charEncoding.equalsIgnoreCase("UTF-8") && !this.charEncoding.equalsIgnoreCase("utf8")) {
                         throw ExceptionFactory.createException(Messages.getString("ServerPreparedStatement.31"), this.session.getExceptionInterceptor());
                     }
                     try {
-                        storeStreamOrReader(parameterIndex, packet, ((Clob) value).getCharacterStream());
+                        storeReader(parameterIndex, packet, ((Clob) value).getCharacterStream(), binding.getScaleOrLength());
                     } catch (Throwable t) {
                         throw ExceptionFactory.createException(t.getMessage(), t);
                     }
@@ -477,88 +477,126 @@ public void setResultFields(ColumnDefinition resultFields) {
         this.resultFields = resultFields;
     }
 
-    private void storeStreamOrReader(int parameterIndex, NativePacketPayload packet, Closeable streamOrReader) {
-        // TODO consider to use more precise type than just Closable
+    private void storeStream(int parameterIndex, NativePacketPayload packet, InputStream inStream, long length) {
         this.session.checkClosed();
-        boolean isStream = InputStream.class.isAssignableFrom(streamOrReader.getClass());
-        byte[] bBuf = null;
-        char[] cBuf = null;
-        String clobEncoding = null;
 
         this.session.getSessionLock().lock();
         try {
-            if (isStream) {
-                bBuf = new byte[BLOB_STREAM_READ_BUF_SIZE];
-            } else {
-                clobEncoding = this.session.getPropertySet().getStringProperty(PropertyKey.clobCharacterEncoding).getStringValue();
-                if (clobEncoding == null) {
-                    clobEncoding = this.session.getPropertySet().getStringProperty(PropertyKey.characterEncoding).getValue();
+            inStream.mark(Integer.MAX_VALUE); // This same stream may have to be read several times, so it must be reset at the end.
+
+            boolean limitLength = length == -1 ? false : this.session.getPropertySet().getBooleanProperty(PropertyKey.useStreamLengthsInPrepStmts).getValue();
+            int sendThreshold = this.session.getPropertySet().getMemorySizeProperty(PropertyKey.blobSendChunkSize).getValue();
+
+            byte[] buffer = new byte[READ_BUFFER_SIZE];
+            boolean dataSent = false;
+            int bytesRead = 0;
+            int bytesInPacket = 0;
+            long lengthLeft = length;
+
+            do {
+                if (bytesInPacket == 0) {
+                    packet.setPosition(0);
+                    this.commandBuilder.buildComStmtSendLongDataHeader(packet, this.serverStatementId, parameterIndex);
+                }
+
+                if (limitLength) {
+                    bytesRead = Util.readBlock(inStream, buffer, lengthLeft, this.session.getExceptionInterceptor());
+                    lengthLeft -= bytesRead;
+                } else {
+                    bytesRead = Util.readBlock(inStream, buffer, this.session.getExceptionInterceptor());
+                }
+                if (bytesRead > 0) {
+                    packet.writeBytes(StringLengthDataType.STRING_FIXED, buffer, 0, bytesRead);
+                    bytesInPacket += bytesRead;
+                }
+
+                if (bytesInPacket >= sendThreshold || bytesRead <= 0 && (!dataSent || bytesInPacket > 0)) {
+                    this.session.getProtocol().sendCommand(packet, true, 0);
+                    dataSent = true;
+                    bytesInPacket = 0;
                 }
-                int maxBytesChar = 2;
-                if (clobEncoding != null) {
-                    maxBytesChar = this.session.getServerSession().getCharsetSettings().getMaxBytesPerChar(clobEncoding);
-                    if (maxBytesChar == 1) {
-                        maxBytesChar = 2; // for safety
+            } while (bytesRead > 0);
+        } finally {
+            try {
+                inStream.reset();
+            } catch (IOException e) {
+            }
+
+            if (this.autoClosePStmtStreams.getValue()) {
+                if (inStream != null) {
+                    try {
+                        inStream.close();
+                    } catch (IOException e) {
                     }
                 }
-                cBuf = new char[ServerPreparedQuery.BLOB_STREAM_READ_BUF_SIZE / maxBytesChar];
             }
 
-            boolean readAny = false;
-            int bytesInPacket = 0;
-            int totalBytesRead = 0;
-            int bytesReadAtLastSend = 0;
-            int packetIsFullAt = this.session.getPropertySet().getMemorySizeProperty(PropertyKey.blobSendChunkSize).getValue();
-            int numRead = 0;
+            this.session.getSessionLock().unlock();
+        }
+    }
 
+    private void storeReader(int parameterIndex, NativePacketPayload packet, Reader reader, long length) {
+        this.session.checkClosed();
+
+        this.session.getSessionLock().lock();
+        try {
             try {
-                packet.setPosition(0);
-                this.commandBuilder.buildComStmtSendLongDataHeader(packet, this.serverStatementId, parameterIndex);
-
-                while ((numRead = isStream ? ((InputStream) streamOrReader).read(bBuf) : ((Reader) streamOrReader).read(cBuf)) != -1) {
-                    readAny = true;
-
-                    if (isStream) {
-                        packet.writeBytes(StringLengthDataType.STRING_FIXED, bBuf, 0, numRead);
-                        bytesInPacket += numRead;
-                        totalBytesRead += numRead;
-                    } else {
-                        byte[] valueAsBytes = StringUtils.getBytes(cBuf, 0, numRead, clobEncoding);
-                        packet.writeBytes(StringSelfDataType.STRING_EOF, valueAsBytes);
-                        bytesInPacket += valueAsBytes.length;
-                        totalBytesRead += valueAsBytes.length;
-                    }
+                reader.mark(Integer.MAX_VALUE); // This same stream may have to be read several times, so it must be reset at the end.
+            } catch (IOException e) {
+            }
 
-                    if (bytesInPacket >= packetIsFullAt) {
-                        bytesReadAtLastSend = totalBytesRead;
+            String clobEncoding = this.session.getPropertySet().getStringProperty(PropertyKey.clobCharacterEncoding).getStringValue();
+            if (clobEncoding == null) {
+                clobEncoding = this.session.getPropertySet().getStringProperty(PropertyKey.characterEncoding).getValue();
+            }
+            boolean limitLength = length == -1 ? false : this.session.getPropertySet().getBooleanProperty(PropertyKey.useStreamLengthsInPrepStmts).getValue();
+            int sendThreshold = this.session.getPropertySet().getMemorySizeProperty(PropertyKey.blobSendChunkSize).getValue();
 
-                        this.session.getProtocol().sendCommand(packet, true, 0);
+            char[] buffer = new char[READ_BUFFER_SIZE];
+            boolean dataSent = false;
+            int charsRead = 0;
+            int bytesInPacket = 0;
+            long lengthLeft = length;
 
-                        bytesInPacket = 0;
-                        packet.setPosition(0);
-                        this.commandBuilder.buildComStmtSendLongDataHeader(packet, this.serverStatementId, parameterIndex);
-                    }
+            do {
+                if (bytesInPacket == 0) {
+                    packet.setPosition(0);
+                    this.commandBuilder.buildComStmtSendLongDataHeader(packet, this.serverStatementId, parameterIndex);
+                }
+
+                if (limitLength) {
+                    charsRead = Util.readBlock(reader, buffer, lengthLeft, this.session.getExceptionInterceptor());
+                    lengthLeft -= charsRead;
+                } else {
+                    charsRead = Util.readBlock(reader, buffer, this.session.getExceptionInterceptor());
+                }
+                if (charsRead > 0) {
+                    byte[] charsAsBytes = StringUtils.getBytes(buffer, 0, charsRead, clobEncoding);
+                    packet.writeBytes(StringSelfDataType.STRING_EOF, charsAsBytes);
+                    bytesInPacket += charsAsBytes.length;
                 }
 
-                if (!readAny || totalBytesRead != bytesReadAtLastSend) {
+                if (bytesInPacket >= sendThreshold || charsRead <= 0 && (!dataSent || bytesInPacket > 0)) {
                     this.session.getProtocol().sendCommand(packet, true, 0);
+                    dataSent = true;
+                    bytesInPacket = 0;
                 }
-            } catch (IOException ioEx) {
-                throw ExceptionFactory.createException(
-                        (isStream ? Messages.getString("ServerPreparedStatement.24") : Messages.getString("ServerPreparedStatement.25")) + ioEx.toString(),
-                        ioEx, this.session.getExceptionInterceptor());
-            } finally {
-                if (this.autoClosePStmtStreams.getValue()) {
-                    if (streamOrReader != null) {
-                        try {
-                            streamOrReader.close();
-                        } catch (IOException ioEx) {
-                            // ignore
-                        }
+            } while (charsRead > 0);
+        } finally {
+            try {
+                reader.reset();
+            } catch (IOException e) {
+            }
+
+            if (this.autoClosePStmtStreams.getValue()) {
+                if (reader != null) {
+                    try {
+                        reader.close();
+                    } catch (IOException ioEx) {
                     }
                 }
             }
-        } finally {
+
             this.session.getSessionLock().unlock();
         }
     }

src/main/protocol-impl/java/com/mysql/cj/protocol/a/InputStreamValueEncoder.java

@@ -61,14 +61,14 @@ public void encodeAsBinary(Message msg, BindValue binding) {
 
     protected byte[] streamToBytes(InputStream in, long length, NativePacketPayload packet) {
         boolean useLength = length == -1 ? false : this.propertySet.getBooleanProperty(PropertyKey.useStreamLengthsInPrepStmts).getValue();
-        in.mark(Integer.MAX_VALUE); // We may need to read this same stream several times, so we need to reset it at the end.
+        in.mark(Integer.MAX_VALUE); // This same stream may have to be read several times, so it must be reset at the end.
         try {
             if (this.streamConvertBuf == null) {
                 this.streamConvertBuf = new byte[4096];
             }
-            int bcnt = useLength ? Util.readBlock(in, this.streamConvertBuf, (int) length, this.exceptionInterceptor)
+            int bcnt = useLength ? Util.readBlock(in, this.streamConvertBuf, length, this.exceptionInterceptor)
                     : Util.readBlock(in, this.streamConvertBuf, this.exceptionInterceptor);
-            int lengthLeftToRead = (int) (length - bcnt);
+            long lengthLeftToRead = length - bcnt;
 
             ByteArrayOutputStream bytesOut = null;
             if (packet == null) {