Fix an issue with storing Unicode surrogate pairs in CLOB columns.

h2/src/main/org/h2/store/LobStorageBackend.java

@@ -9,6 +9,10 @@ package org.h2.store;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.Reader;
+import java.nio.ByteBuffer;
+import java.nio.CharBuffer;
+import java.nio.charset.CharsetEncoder;
+import java.nio.charset.CodingErrorAction;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
@@ -16,6 +20,7 @@ import java.sql.Statement;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
+
 import org.h2.constant.ErrorCode;
 import org.h2.constant.SysProperties;
 import org.h2.engine.Constants;
@@ -26,7 +31,6 @@ import org.h2.tools.CompressTool;
 import org.h2.util.IOUtils;
 import org.h2.util.MathUtils;
 import org.h2.util.New;
-import org.h2.util.Utils;
 import org.h2.value.Value;
 import org.h2.value.ValueLob;
 import org.h2.value.ValueLobDb;
@@ -41,9 +45,10 @@ import org.h2.value.ValueLobDb;
  * take a very long time. If we did them on a normal session, we would be
  * locking the LOB tables for long periods of time, which is extremely
  * detrimental to the rest of the system. Perhaps when we shift to the MVStore
- * engine, we can revisit this design decision.
+ * engine, we can revisit this design decision 
+ * (using the StreamStore, that is, no connection at all).
  * <p>
- * Locking Discussion
+ * Locking
  * <p>
  * Normally, the locking order in H2 is: first lock the Session object, then
  * lock the Database object. However, in the case of the LOB data, we are using
@@ -67,10 +72,6 @@ import org.h2.value.ValueLobDb;
  */
 public class LobStorageBackend implements LobStorageInterface {
 
-    /**
-     * Locking Discussion
-     * --------------------
-     */
     /**
      * The name of the lob data table. If this table exists, then lob storage is
      * used.
@@ -782,76 +783,88 @@ public class LobStorageBackend implements LobStorageInterface {
         }
 
     }
-
+    
     /**
      * An input stream that reads the data from a reader.
      */
-    static class CountingReaderInputStream extends InputStream {
+    public static class CountingReaderInputStream extends InputStream {
 
         private final Reader reader;
-        /**
-         * total length of Reader data in chars
-         */
+        
+        private final CharBuffer charBuffer = CharBuffer.allocate(Constants.IO_BUFFER_SIZE);
+        
+        private final CharsetEncoder encoder = Constants.UTF8.newEncoder().
+                onMalformedInput(CodingErrorAction.REPLACE).
+                onUnmappableCharacter(CodingErrorAction.REPLACE);
+        
+        private ByteBuffer byteBuffer = ByteBuffer.allocate(0);
         private long length;
         private long remaining;
-        private int pos;
-        private final char[] charBuffer = new char[Constants.IO_BUFFER_SIZE];
-        private byte[] buffer;
 
         CountingReaderInputStream(Reader reader, long maxLength) {
             this.reader = reader;
             this.remaining = maxLength;
-            buffer = Utils.EMPTY_BYTES;
         }
 
         @Override
         public int read(byte[] buff, int offset, int len) throws IOException {
-            if (buffer == null) {
+            if (!fetch()) {
                 return -1;
             }
-            if (pos >= buffer.length) {
-                fillBuffer();
-                if (buffer == null) {
-                    return -1;
-                }
-            }
-            len = Math.min(len, buffer.length - pos);
-            System.arraycopy(buffer, pos, buff, offset, len);
-            pos += len;
+            len = Math.min(len, byteBuffer.remaining());
+            byteBuffer.get(buff, offset, len);
             return len;
         }
 
         @Override
         public int read() throws IOException {
-            if (buffer == null) {
+            if (!fetch()) {
                 return -1;
             }
-            if (pos >= buffer.length) {
+            return byteBuffer.get() & 255;
+        }
+        
+        private boolean fetch() throws IOException {
+            if (byteBuffer != null && byteBuffer.remaining() == 0) {
                 fillBuffer();
-                if (buffer == null) {
-                    return -1;
-                }
             }
-            return buffer[pos++];
+            return byteBuffer != null;
         }
 
         private void fillBuffer() throws IOException {
-            int len = (int) Math.min(charBuffer.length, remaining);
+            int len = (int) Math.min(charBuffer.capacity() - charBuffer.position(), remaining);
             if (len > 0) {
-                len = reader.read(charBuffer, 0, len);
-            } else {
-                len = -1;
+                len = reader.read(charBuffer.array(), charBuffer.position(), len);
             }
-            if (len < 0) {
-                buffer = null;
-            } else {
-                buffer = new String(charBuffer, 0, len).getBytes(Constants.UTF8);
-                length += len;
+            if (len > 0) {
                 remaining -= len;
+            } else {
+                len = 0;
+                remaining = 0;
+            }
+            length += len;
+            charBuffer.limit(charBuffer.position() + len);
+            charBuffer.rewind();
+            byteBuffer = ByteBuffer.allocate(Constants.IO_BUFFER_SIZE);
+            boolean end = remaining == 0;
+            encoder.encode(charBuffer, byteBuffer, end);
+            if (end && byteBuffer.position() == 0) {
+                // EOF
+                byteBuffer = null;
+                return;
             }
-            pos = 0;
+            byteBuffer.flip();
+            charBuffer.compact();
+            charBuffer.flip();
+            charBuffer.position(charBuffer.limit());
         }
-
+        
+        /**
+         * The number of characters read so far (but there might still be some bytes
+         * in the buffer).
+         * 
+         * @return the number of characters
+         */
         public long getLength() {
             return length;
         }

h2/src/main/org/h2/value/ValueLobDb.java

@@ -7,10 +7,10 @@
 package org.h2.value;
 
 import java.io.BufferedInputStream;
+import java.io.BufferedReader;
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
-import java.io.InputStreamReader;
 import java.io.Reader;
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
@@ -86,11 +86,11 @@ public class ValueLobDb extends Value implements Value.ValueClob, Value.ValueBlo
         this.fileName = null;
         this.tempFile = null;
     }
-
+    
     /**
      * Create temporary CLOB from Reader.
      */
-    private ValueLobDb(DataHandler handler, char[] buff, int len, Reader in, long remaining) throws IOException {
+    private ValueLobDb(DataHandler handler, Reader in, long remaining) throws IOException {
         this.type = Value.CLOB;
         this.handler = handler;
         this.small = null;
@@ -102,15 +102,9 @@ public class ValueLobDb extends Value implements Value.ValueClob, Value.ValueBlo
         FileStoreOutputStream out = new FileStoreOutputStream(tempFile, null, null);
         long tmpPrecision = 0;
         try {
+            char[] buff = new char[Constants.IO_BUFFER_SIZE];
             while (true) {
-                tmpPrecision += len;
-                byte[] b = new String(buff, 0, len).getBytes(Constants.UTF8);
-                out.write(b, 0, b.length);
-                remaining -= len;
-                if (remaining <= 0) {
-                    break;
-                }
-                len = getBufferSize(this.handler, false, remaining);
+                int len = getBufferSize(this.handler, false, remaining);
                 len = IOUtils.readFully(in, buff, len);
                 if (len <= 0) {
                     break;
@@ -392,7 +386,7 @@ public class ValueLobDb extends Value implements Value.ValueClob, Value.ValueBlo
 
     @Override
     public Reader getReader() {
-        return new InputStreamReader(getInputStream(), Constants.UTF8);
+        return IOUtils.getBufferedReader(getInputStream());
     }
 
     @Override
@@ -510,6 +504,12 @@ public class ValueLobDb extends Value implements Value.ValueClob, Value.ValueBlo
      * @return the lob value
      */
     public static ValueLobDb createTempClob(Reader in, long length, DataHandler handler) {
+        BufferedReader reader;
+        if (in instanceof BufferedReader) {
+            reader = (BufferedReader) in;
+        } else {
+            reader = new BufferedReader(in, Constants.IO_BUFFER_SIZE);
+        }
         try {
             boolean compress = handler.getLobCompressionAlgorithm(Value.CLOB) != null;
             long remaining = Long.MAX_VALUE;
@@ -519,19 +519,21 @@ public class ValueLobDb extends Value implements Value.ValueClob, Value.ValueBlo
             int len = getBufferSize(handler, compress, remaining);
             char[] buff;
             if (len >= Integer.MAX_VALUE) {
-                String data = IOUtils.readStringAndClose(in, -1);
+                String data = IOUtils.readStringAndClose(reader, -1);
                 buff = data.toCharArray();
                 len = buff.length;
             } else {
                 buff = new char[len];
-                len = IOUtils.readFully(in, buff, len);
+                reader.mark(len);
+                len = IOUtils.readFully(reader, buff, len);
                 len = len < 0 ? 0 : len;
             }
             if (len <= handler.getMaxLengthInplaceLob()) {
                 byte[] small = new String(buff, 0, len).getBytes(Constants.UTF8);
                 return ValueLobDb.createSmallLob(Value.CLOB, small, len);
             }
-            ValueLobDb lob = new ValueLobDb(handler, buff, len, in, remaining);
+            reader.reset();
+            ValueLobDb lob = new ValueLobDb(handler, reader, remaining);
             return lob;
         } catch (IOException e) {
             throw DbException.convertIOException(e, null);

h2/src/test/org/h2/test/db/TestLob.java

@@ -64,6 +64,7 @@ public class TestLob extends TestBase {
 
     @Override
     public void test() throws Exception {
+        testClobWithRandomUnicodeChars();
         testCommitOnExclusiveConnection();
         testReadManyLobs();
         testLobSkip();
@@ -113,7 +114,6 @@ public class TestLob extends TestBase {
         testLob(false);
         testLob(true);
         testJavaObject();
-        testClobWithRandomUnicodeChars();
         deleteDb("lob");
         FileUtils.deleteRecursive(TEMP_DIR, true);
     }
@@ -1517,7 +1517,7 @@ public class TestLob extends TestBase {
         Statement stat = conn.createStatement();
         stat.execute("CREATE TABLE logs(id int primary key auto_increment, message CLOB)");
         PreparedStatement s1 = conn.prepareStatement("INSERT INTO logs (id, message) VALUES(null, ?)");
-        final Random rand = new Random();
+        final Random rand = new Random(1);
         for (int i = 1; i <= 100; i++) {
             String data = randomUnicodeString(rand);
             s1.setString(1, data);
@@ -1525,6 +1525,11 @@ public class TestLob extends TestBase {
             ResultSet rs = stat.executeQuery("SELECT id, message FROM logs ORDER BY id DESC LIMIT 1");
             rs.next();
             String read = rs.getString(2);
+            if (!read.equals(data)) {
+                for (int j = 0; j < read.length(); j++) {
+                    assertEquals("pos: " + j, read.charAt(j), data.charAt(j));
+                }
+            }
             assertEquals(read, data);
         }
         conn.close();