Implement JdbcBlob.getBinaryStream() and JdbcClob.getCharacterStream() with pos…

Implement JdbcBlob.getBinaryStream() and JdbcClob.getCharacterStream() with pos and length arguments

h2/src/main/org/h2/jdbc/JdbcBlob.java

@@ -304,7 +304,7 @@ public class JdbcBlob extends TraceObject implements Blob {
     }
 
     /**
-     * [Not supported] Returns the input stream, starting from an offset.
+     * Returns the input stream, starting from an offset.
      *
      * @param pos where to start reading
      * @param length the number of bytes that will be read
@@ -312,7 +312,13 @@ public class JdbcBlob extends TraceObject implements Blob {
      */
     @Override
     public InputStream getBinaryStream(long pos, long length) throws SQLException {
-        throw unsupported("LOB update");
+        try {
+            debugCodeCall("getBinaryStream(pos, length)");
+            checkClosed();
+            return value.getInputStream(pos, length);
+        } catch (Exception e) {
+            throw logAndConvert(e);
+        }
     }
 
     private void checkClosed() {

h2/src/main/org/h2/jdbc/JdbcClob.java

@@ -261,11 +261,17 @@ public class JdbcClob extends TraceObject implements NClob
     }
 
     /**
-     * [Not supported] Returns the reader, starting from an offset.
+     * Returns the reader, starting from an offset.
      */
     @Override
     public Reader getCharacterStream(long pos, long length) throws SQLException {
-        throw unsupported("LOB subset");
+        try {
+            debugCodeCall("getCharacterStream(pos, length)");
+            checkClosed();
+            return value.getReader(pos, length);
+        } catch (Exception e) {
+            throw logAndConvert(e);
+        }
     }
 
     private void checkClosed() {

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

+package org.h2.store;
+
+import java.io.FilterInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+
+public final class RangeInputStream extends FilterInputStream {
+    private long offset, limit;
+
+    public RangeInputStream(InputStream in, long offset, long limit) {
+        super(in);
+        this.offset = offset;
+        this.limit = limit;
+    }
+
+    private void before() throws IOException {
+        while (offset > 0) {
+            offset -= in.skip(offset);
+        }
+    }
+
+    @Override
+    public int read() throws IOException {
+        before();
+        if (limit < 1) {
+            return -1;
+        }
+        int b = in.read();
+        if (b >= 0) {
+            limit--;
+        }
+        return b;
+    }
+
+    @Override
+    public int read(byte b[], int off, int len) throws IOException {
+        before();
+        if (len > limit) {
+            len = (int) limit;
+        }
+        int cnt = in.read(b, off, len);
+        if (cnt > 0) {
+            limit -= cnt;
+        }
+        return cnt;
+    }
+
+    @Override
+    public long skip(long n) throws IOException {
+        before();
+        if (n > limit) {
+            n = (int) limit;
+        }
+        n = in.skip(n);
+        limit -= n;
+        return n;
+    }
+
+    @Override
+    public int available() throws IOException {
+        before();
+        int cnt = in.available();
+        if (cnt > limit) {
+            return (int) limit;
+        }
+        return cnt;
+    }
+
+    @Override
+    public void close() throws IOException {
+        in.close();
+    }
+
+    @Override
+    public void mark(int readlimit) {
+    }
+
+    @Override
+    public synchronized void reset() throws IOException {
+        throw new IOException("mark/reset not supported");
+    }
+
+    @Override
+    public boolean markSupported() {
+        return false;
+    }
+}

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

+package org.h2.store;
+
+import java.io.IOException;
+import java.io.Reader;
+
+public final class RangeReader extends Reader {
+    private final Reader r;
+
+    private long offset, limit;
+
+    public RangeReader(Reader r, long offset, long limit) {
+        this.r = r;
+        this.offset = offset;
+        this.limit = limit;
+    }
+
+    private void before() throws IOException {
+        while (offset > 0) {
+            offset -= r.skip(offset);
+        }
+    }
+
+    @Override
+    public int read() throws IOException {
+        before();
+        if (limit < 1) {
+            return -1;
+        }
+        int c = r.read();
+        if (c >= 0) {
+            limit--;
+        }
+        return c;
+    }
+
+    @Override
+    public int read(char cbuf[], int off, int len) throws IOException {
+        before();
+        if (len > limit) {
+            len = (int) limit;
+        }
+        int cnt = r.read(cbuf, off, len);
+        if (cnt > 0) {
+            limit -= cnt;
+        }
+        return cnt;
+    }
+
+    @Override
+    public long skip(long n) throws IOException {
+        before();
+        if (n > limit) {
+            n = (int) limit;
+        }
+        n = r.skip(n);
+        limit -= n;
+        return n;
+    }
+
+    @Override
+    public boolean ready() throws IOException {
+        before();
+        if (limit > 0) {
+            return r.ready();
+        }
+        return false;
+    }
+
+    @Override
+    public boolean markSupported() {
+        return false;
+    }
+
+    @Override
+    public void mark(int readAheadLimit) throws IOException {
+        throw new IOException("mark() not supported");
+    }
+
+    @Override
+    public void reset() throws IOException {
+        throw new IOException("reset() not supported");
+    }
+
+    @Override
+    public void close() throws IOException {
+        r.close();
+    }
+}

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

@@ -19,6 +19,7 @@ import java.sql.SQLException;
 import java.sql.Time;
 import java.sql.Timestamp;
 import java.sql.Types;
+
 import org.h2.api.ErrorCode;
 import org.h2.engine.Mode;
 import org.h2.engine.SysProperties;
@@ -187,6 +188,15 @@ public abstract class Value {
     private static final BigDecimal MIN_LONG_DECIMAL =
             BigDecimal.valueOf(Long.MIN_VALUE);
 
+    private static void rangeCheck(long zeroBasedOffset, long length, long dataSize) {
+        if ((zeroBasedOffset | length) < 0 || length > dataSize - zeroBasedOffset) {
+            if (zeroBasedOffset < 0 || zeroBasedOffset > dataSize) {
+                throw DbException.getInvalidValueException("offset", zeroBasedOffset + 1);
+            }
+            throw DbException.getInvalidValueException("length", length);
+        }
+    }
+
     /**
      * Get the SQL expression for this value.
      *
@@ -469,10 +479,23 @@ public abstract class Value {
         return new ByteArrayInputStream(getBytesNoCopy());
     }
 
+    public InputStream getInputStream(long oneBasedOffset, long length) {
+        byte[] bytes = getBytesNoCopy();
+        rangeCheck(--oneBasedOffset, length, bytes.length);
+        return new ByteArrayInputStream(bytes, /* 0-based */ (int) oneBasedOffset, (int) length);
+    }
+
     public Reader getReader() {
         return new StringReader(getString());
     }
 
+    public Reader getReader(long oneBasedOffset, long length) {
+        String string = getString();
+        rangeCheck(--oneBasedOffset, length, string.length());
+        int offset = /* 0-based */ (int) oneBasedOffset;
+        return new StringReader(string.substring(offset, offset + (int) length));
+    }
+
     /**
      * Add a value and return the result.
      *

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

@@ -23,6 +23,8 @@ import org.h2.store.DataHandler;
 import org.h2.store.FileStore;
 import org.h2.store.FileStoreInputStream;
 import org.h2.store.FileStoreOutputStream;
+import org.h2.store.RangeInputStream;
+import org.h2.store.RangeReader;
 import org.h2.store.fs.FileUtils;
 import org.h2.table.Column;
 import org.h2.util.IOUtils;
@@ -48,6 +50,25 @@ import org.h2.util.Utils;
  */
 public class ValueLob extends Value {
 
+    private static void rangeCheckUnknown(long zeroBasedOffset, long length) {
+        if (zeroBasedOffset < 0) {
+            throw DbException.getInvalidValueException("offset", zeroBasedOffset + 1);
+        }
+        if (length < 0) {
+            throw DbException.getInvalidValueException("length", length);
+        }
+    }
+
+    static InputStream rangeInputStream(InputStream inputStream, long oneBasedOffset, long length) {
+        rangeCheckUnknown(--oneBasedOffset, length);
+        return new RangeInputStream(inputStream, /* 0-based */ oneBasedOffset, length);
+    }
+
+    static Reader rangeReader(Reader reader, long oneBasedOffset, long length) {
+        rangeCheckUnknown(--oneBasedOffset, length);
+        return new RangeReader(reader, /* 0-based */ oneBasedOffset, length);
+    }
+
     /**
      * This counter is used to calculate the next directory to store lobs. It is
      * better than using a random number because less directories are created.
@@ -632,6 +653,11 @@ public class ValueLob extends Value {
         return IOUtils.getBufferedReader(getInputStream());
     }
 
+    @Override
+    public Reader getReader(long oneBasedOffset, long length) {
+        return rangeReader(getReader(), oneBasedOffset, length);
+    }
+
     @Override
     public InputStream getInputStream() {
         if (fileName == null) {
@@ -644,6 +670,14 @@ public class ValueLob extends Value {
                 Constants.IO_BUFFER_SIZE);
     }
 
+    @Override
+    public InputStream getInputStream(long oneBasedOffset, long length) {
+        if (fileName == null) {
+            return super.getInputStream(oneBasedOffset, length);
+        }
+        return rangeInputStream(getInputStream(), oneBasedOffset, length);
+    }
+
     @Override
     public void set(PreparedStatement prep, int parameterIndex)
             throws SQLException {

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

@@ -374,6 +374,11 @@ public class ValueLobDb extends Value implements Value.ValueClob,
         return IOUtils.getBufferedReader(getInputStream());
     }
 
+    @Override
+    public Reader getReader(long oneBasedOffset, long length) {
+        return ValueLob.rangeReader(getReader(), oneBasedOffset, length);
+    }
+
     @Override
     public InputStream getInputStream() {
         if (small != null) {
@@ -392,6 +397,14 @@ public class ValueLobDb extends Value implements Value.ValueClob,
         }
     }
 
+    @Override
+    public InputStream getInputStream(long oneBasedOffset, long length) {
+        if (small != null) {
+            return super.getInputStream(oneBasedOffset, length);
+        }
+        return ValueLob.rangeInputStream(getInputStream(), oneBasedOffset, length);
+    }
+
     @Override
     public void set(PreparedStatement prep, int parameterIndex)
             throws SQLException {

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

@@ -1074,18 +1074,34 @@ public class TestLob extends TestBase {
         prep2.getQueryTimeout();
         prep2.close();
         conn0.getAutoCommit();
-        Reader r = clob0.getCharacterStream();
+        Reader r;
+        int ch;
+        r = clob0.getCharacterStream();
         for (int i = 0; i < 10000; i++) {
-            int ch = r.read();
+            ch = r.read();
             if (ch != ('0' + (i % 10))) {
                 fail("expected " + (char) ('0' + (i % 10)) +
                         " got: " + ch + " (" + (char) ch + ")");
             }
         }
-        int ch = r.read();
+        ch = r.read();
         if (ch != -1) {
             fail("expected -1 got: " + ch);
         }
+        r.close();
+        r = clob0.getCharacterStream(1235, 1000);
+        for (int i = 1234; i < 2234; i++) {
+            ch = r.read();
+            if (ch != ('0' + (i % 10))) {
+                fail("expected " + (char) ('0' + (i % 10)) +
+                        " got: " + ch + " (" + (char) ch + ")");
+            }
+        }
+        ch = r.read();
+        if (ch != -1) {
+            fail("expected -1 got: " + ch);
+        }
+        r.close();
         conn0.close();
     }
 

h2/src/test/org/h2/test/jdbc/TestResultSet.java

@@ -32,6 +32,7 @@ import java.sql.Statement;
 import java.sql.Time;
 import java.sql.Timestamp;
 import java.sql.Types;
+import java.util.Arrays;
 import java.util.Calendar;
 import java.util.Collections;
 import java.util.TimeZone;
@@ -42,6 +43,8 @@ import org.h2.test.TestBase;
 import org.h2.util.DateTimeUtils;
 import org.h2.util.IOUtils;
 import org.h2.util.LocalDateTimeUtils;
+import org.h2.util.MathUtils;
+import org.h2.util.StringUtils;
 
 /**
  * Tests for the ResultSet implementation.
@@ -1548,6 +1551,9 @@ public class TestResultSet extends TestBase {
         stat.execute("INSERT INTO TEST VALUES(5,X'0bcec1')");
         stat.execute("INSERT INTO TEST VALUES(6,X'03030303')");
         stat.execute("INSERT INTO TEST VALUES(7,NULL)");
+        byte[] random = new byte[0x10000];
+        MathUtils.randomBytes(random);
+        stat.execute("INSERT INTO TEST VALUES(8, X'" + StringUtils.convertBytesToHex(random) + "')");
         rs = stat.executeQuery("SELECT * FROM TEST ORDER BY ID");
         assertResultSetMeta(rs, 2, new String[] { "ID", "VALUE" },
                 new int[] { Types.INTEGER, Types.BLOB }, new int[] {
@@ -1587,14 +1593,27 @@ public class TestResultSet extends TestBase {
         InputStream in = rs.getBinaryStream("value");
         byte[] b = readAllBytes(in);
         assertEqualsWithNull(new byte[] { (byte) 0x0b, (byte) 0xce, (byte) 0xc1 }, b);
+        Blob blob = rs.getObject("value", Blob.class);
+        try {
+            assertTrue(blob != null);
+            assertEqualsWithNull(new byte[] { (byte) 0x0b, (byte) 0xce, (byte) 0xc1 },
+                    readAllBytes(blob.getBinaryStream()));
+            assertEqualsWithNull(new byte[] { (byte) 0xce,
+                    (byte) 0xc1 }, readAllBytes(blob.getBinaryStream(2, 2)));
+            assertTrue(!rs.wasNull());
+        } finally {
+            blob.free();
+        }
         assertTrue(!rs.wasNull());
         rs.next();
 
-        Blob blob = rs.getObject("value", Blob.class);
+        blob = rs.getObject("value", Blob.class);
         try {
             assertTrue(blob != null);
             assertEqualsWithNull(new byte[] { (byte) 0x03, (byte) 0x03,
                     (byte) 0x03, (byte) 0x03 }, readAllBytes(blob.getBinaryStream()));
+            assertEqualsWithNull(new byte[] { (byte) 0x03,
+                    (byte) 0x03 }, readAllBytes(blob.getBinaryStream(2, 2)));
             assertTrue(!rs.wasNull());
         } finally {
             blob.free();
@@ -1603,6 +1622,20 @@ public class TestResultSet extends TestBase {
 
         assertEqualsWithNull(null, readAllBytes(rs.getBinaryStream("VaLuE")));
         assertTrue(rs.wasNull());
+        rs.next();
+
+        blob = rs.getObject("value", Blob.class);
+        try {
+            assertTrue(blob != null);
+            assertEqualsWithNull(random, readAllBytes(blob.getBinaryStream()));
+            byte[] expected = Arrays.copyOfRange(random, 100, 50102);
+            byte[] got = readAllBytes(blob.getBinaryStream(101, 50002));
+            assertEqualsWithNull(expected, got);
+            assertTrue(!rs.wasNull());
+        } finally {
+            blob.free();
+        }
+
         assertTrue(!rs.next());
         stat.execute("DROP TABLE TEST");
     }
@@ -1659,6 +1692,12 @@ public class TestResultSet extends TestBase {
         assertTrue(!rs.wasNull());
         trace(string);
         assertTrue(string != null && string.equals("Hallo"));
+        Clob clob = rs.getClob(2);
+        try {
+            assertEquals("all", readString(clob.getCharacterStream(2, 3)));
+        } finally {
+            clob.free();
+        }
         rs.next();
 
         string = readString(rs.getCharacterStream("value"));
@@ -1667,7 +1706,7 @@ public class TestResultSet extends TestBase {
         assertTrue(string != null && string.equals("Welt!"));
         rs.next();
 
-        Clob clob = rs.getObject("value", Clob.class);
+        clob = rs.getObject("value", Clob.class);
         try {
             assertTrue(clob != null);
             string = readString(clob.getCharacterStream());