OPENJPA-130. Committing Ignacio's patch, with a few whitespace and method name tweaks.

git-svn-id: https://svn.apache.org/repos/asf/openjpa/trunk@574143 13f79535-47bb-0310-9956-ffa450edef68

openjpa-jdbc/src/main/java/org/apache/openjpa/jdbc/meta/MappingRepository.java

@@ -40,6 +40,7 @@ import org.apache.openjpa.jdbc.meta.strats.FlatClassStrategy;
 import org.apache.openjpa.jdbc.meta.strats.FullClassStrategy;
 import org.apache.openjpa.jdbc.meta.strats.HandlerFieldStrategy;
 import org.apache.openjpa.jdbc.meta.strats.ImmutableValueHandler;
+import org.apache.openjpa.jdbc.meta.strats.LobFieldStrategy;
 import org.apache.openjpa.jdbc.meta.strats.MaxEmbeddedBlobFieldStrategy;
 import org.apache.openjpa.jdbc.meta.strats.MaxEmbeddedByteArrayFieldStrategy;
 import org.apache.openjpa.jdbc.meta.strats.MaxEmbeddedCharArrayFieldStrategy;
@@ -866,6 +867,9 @@ public class MappingRepository
                     break;
                 return handlerMapStrategy(field, khandler, vhandler, krel,
                     vrel, installHandlers);
+            case JavaTypes.INPUT_STREAM:
+            case JavaTypes.INPUT_READER:
+                return new LobFieldStrategy();
         }
         return null;
     }

openjpa-jdbc/src/main/java/org/apache/openjpa/jdbc/meta/strats/LobFieldStrategy.java

@@ -0,0 +1,194 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.    
+ */
+package org.apache.openjpa.jdbc.meta.strats;
+
+import java.io.InputStream;
+import java.io.Reader;
+import java.sql.SQLException;
+import java.sql.Types;
+
+import org.apache.openjpa.jdbc.kernel.JDBCFetchConfiguration;
+import org.apache.openjpa.jdbc.kernel.JDBCStore;
+import org.apache.openjpa.jdbc.meta.FieldMapping;
+import org.apache.openjpa.jdbc.meta.ValueMappingInfo;
+import org.apache.openjpa.jdbc.schema.Column;
+import org.apache.openjpa.jdbc.sql.Result;
+import org.apache.openjpa.jdbc.sql.Row;
+import org.apache.openjpa.jdbc.sql.RowManager;
+import org.apache.openjpa.jdbc.sql.Select;
+import org.apache.openjpa.kernel.OpenJPAStateManager;
+
+/**
+ * Direct mapping from a stream value to a column.
+ *
+ * @author Ignacio Andreu
+ * @since 1.1.0
+ */
+public class LobFieldStrategy extends AbstractFieldStrategy {
+
+    private int fieldType;
+
+    public void map(boolean adapt) {
+        assertNotMappedBy();
+        field.mapJoin(adapt, false);
+        field.getKeyMapping().getValueInfo().assertNoSchemaComponents
+            (field.getKey(), !adapt);
+        field.getElementMapping().getValueInfo().assertNoSchemaComponents
+            (field.getElement(), !adapt);
+        field.setStream(true);
+        ValueMappingInfo vinfo = field.getValueInfo();
+        vinfo.assertNoJoin(field, true);
+        vinfo.assertNoForeignKey(field, !adapt);
+        Column tmpCol = new Column();
+        tmpCol.setName(field.getName());
+        tmpCol.setJavaType(field.getTypeCode());
+        tmpCol.setType(fieldType);
+        tmpCol.setSize(-1);
+
+        Column[] cols = vinfo.getColumns(field, field.getName(),
+            new Column[]{ tmpCol }, field.getTable(), adapt);
+
+        field.setColumns(cols);
+        field.setColumnIO(vinfo.getColumnIO());
+        field.mapConstraints(field.getName(), adapt);
+        field.mapPrimaryKey(adapt);
+    }
+
+    public Boolean isCustomInsert(OpenJPAStateManager sm, JDBCStore store) {
+        return null;
+    }
+
+    public void insert(OpenJPAStateManager sm, JDBCStore store, RowManager rm)
+        throws SQLException {
+        Object ob = toDataStoreValue(sm.fetchObjectField
+            (field.getIndex()), store);
+        Row row = field.getRow(sm, store, rm, Row.ACTION_INSERT);
+        if (field.getColumnIO().isInsertable(0, ob == null)) {
+            if (ob != null) {
+                if (isBlob()) {
+                    store.getDBDictionary().insertBlobForStreamingLoad
+                        (row, field.getColumns()[0]);
+                } else {
+                    store.getDBDictionary().insertClobForStreamingLoad
+                        (row, field.getColumns()[0]);
+                }
+            } else {
+                Column col = field.getColumns()[0];
+                col.setType(Types.OTHER);
+                row.setNull(col);
+            }
+        }
+    }
+
+    public void customInsert(OpenJPAStateManager sm, JDBCStore store)
+        throws SQLException {
+        Object ob = toDataStoreValue(sm.fetchObjectField
+            (field.getIndex()), store);
+        if (field.getColumnIO().isInsertable(0, ob == null)) {
+            if (ob != null) {
+                Select sel = createSelect(sm, store);
+                if (isBlob()) {
+                    store.getDBDictionary().updateBlob
+                        (sel, store, (InputStream)ob);
+                } else {
+                    store.getDBDictionary().updateClob
+                        (sel, store, (Reader)ob);
+                }
+            }
+        }
+    }
+
+    public void update(OpenJPAStateManager sm, JDBCStore store, RowManager rm)
+        throws SQLException {
+        Object ob = toDataStoreValue(sm.fetchObjectField
+            (field.getIndex()), store);
+        if (field.getColumnIO().isUpdatable(0, ob == null)) {
+            if (ob != null) {
+                Select sel = createSelect(sm, store);
+                if (isBlob()) {
+                    store.getDBDictionary().updateBlob
+                        (sel, store, (InputStream)ob);
+                } else {
+                    store.getDBDictionary().updateClob
+                        (sel, store, (Reader)ob);
+                }
+            } else {
+                Row row = field.getRow(sm, store, rm, Row.ACTION_UPDATE);
+                Column col = field.getColumns()[0];
+                col.setType(Types.OTHER);
+                row.setNull(col);
+            }
+        }
+    }
+
+    public int supportsSelect(Select sel, int type, OpenJPAStateManager sm,
+        JDBCStore store, JDBCFetchConfiguration fetch) {
+        if (type == Select.TYPE_JOINLESS && sel.isSelected(field.getTable()))
+            return 1;
+        return 0;
+    }
+
+    public int select(Select sel, OpenJPAStateManager sm, JDBCStore store,
+        JDBCFetchConfiguration fetch, int eagerMode) {
+        sel.select(field.getColumns()[0], field.join(sel));
+        return 1;
+    }
+
+    public void load(OpenJPAStateManager sm, JDBCStore store,
+        JDBCFetchConfiguration fetch, Result res) throws SQLException {
+        Column col = field.getColumns()[0];
+        if (res.contains(col)) {
+            if (isBlob()) {
+                sm.storeObject(field.getIndex(), res.getBinaryStream(col));
+            } else {
+                sm.storeObject(field.getIndex(), res.getCharacterStream(col));
+            }
+        }
+    }
+
+    protected void assertNotMappedBy() {
+        if (field != null && field.getMappedBy() != null)
+            throw new UnsupportedOperationException();
+    }
+
+    public void setFieldMapping(FieldMapping owner) {
+        if (owner.getType().isAssignableFrom(InputStream.class)) {
+            fieldType = Types.BLOB;
+        } else if (owner.getType().isAssignableFrom(Reader.class)) {
+            fieldType = Types.CLOB;
+        }
+        field = owner;
+    }
+
+    private boolean isBlob() {
+        if (fieldType == Types.BLOB)
+            return true;
+        return false;
+    }
+
+    private Select createSelect(OpenJPAStateManager sm, JDBCStore store) {
+        Select sel = store.getSQLFactory().newSelect();
+        sel.select(field.getColumns()[0]);
+        sel.selectPrimaryKey(field.getDefiningMapping());
+        sel.wherePrimaryKey
+            (sm.getObjectId(), field.getDefiningMapping(), store);
+        sel.setLob(true);
+        return sel;
+    }
+}

openjpa-jdbc/src/main/java/org/apache/openjpa/jdbc/sql/DBDictionary.java

@@ -20,9 +20,11 @@ package org.apache.openjpa.jdbc.sql;
 
 import java.io.BufferedReader;
 import java.io.ByteArrayInputStream;
+import java.io.CharArrayReader;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
+import java.io.OutputStream;
 import java.io.Reader;
 import java.io.StringReader;
 import java.io.Writer;
@@ -96,6 +98,7 @@ import org.apache.openjpa.lib.util.Localizer;
 import org.apache.openjpa.lib.util.Localizer.Message;
 import org.apache.openjpa.meta.JavaTypes;
 import org.apache.openjpa.util.GeneralException;
+import org.apache.openjpa.util.InternalException;
 import org.apache.openjpa.util.OpenJPAException;
 import org.apache.openjpa.util.ReferentialIntegrityException;
 import org.apache.openjpa.util.Serialization;
@@ -125,6 +128,9 @@ public class DBDictionary
     public static final String CONS_NAME_MID = "mid";
     public static final String CONS_NAME_AFTER = "after";
     
+    public int blobBufferSize = 50;
+    public int clobBufferSize = 50;
+
     protected static final int RANGE_POST_SELECT = 0;
     protected static final int RANGE_PRE_DISTINCT = 1;
     protected static final int RANGE_POST_DISTINCT = 2;
@@ -3920,6 +3926,109 @@ public class DBDictionary
         return column.toString();
     }
     
+    public void insertBlobForStreamingLoad(Row row, Column col)
+    throws SQLException {
+        row.setBinaryStream(col, 
+                new ByteArrayInputStream(new byte[0]), 0);
+    }
+    
+    public void insertClobForStreamingLoad(Row row, Column col)
+    throws SQLException {
+        row.setCharacterStream(col,
+                new CharArrayReader(new char[0]), 0);
+    }
+    
+    public void updateBlob(Select sel, JDBCStore store, InputStream is)
+        throws SQLException {
+        SQLBuffer sql = sel.toSelect(true, store.getFetchConfiguration());
+        ResultSet res = null;
+        Connection conn = store.getConnection();
+        PreparedStatement stmnt = null;
+        try {
+            stmnt = sql.prepareStatement(conn, store.getFetchConfiguration(),
+                ResultSet.TYPE_SCROLL_SENSITIVE,
+                ResultSet.CONCUR_UPDATABLE);
+            res = stmnt.executeQuery();
+            if (!res.next()) {
+                throw new InternalException(_loc.get("stream-exception"));
+            }
+            Blob blob = res.getBlob(1);
+            OutputStream os = blob.setBinaryStream(1);
+            copy(is, os);
+            os.close();
+            res.updateBlob(1, blob);
+            res.updateRow();
+
+        } catch (IOException ioe) {
+            throw new StoreException(ioe);
+        } finally {
+            if (res != null)
+                try { res.close (); } catch (SQLException e) {}
+            if (stmnt != null)
+                try { stmnt.close (); } catch (SQLException e) {}
+            if (conn != null)
+                try { conn.close (); } catch (SQLException e) {}
+        }
+    }
+    
+    public void updateClob(Select sel, JDBCStore store, Reader reader)
+        throws SQLException {
+        SQLBuffer sql = sel.toSelect(true, store.getFetchConfiguration());
+        ResultSet res = null;
+        Connection conn = store.getConnection();
+        PreparedStatement stmnt = null;
+        try {
+            stmnt = sql.prepareStatement(conn, store.getFetchConfiguration(),
+                ResultSet.TYPE_SCROLL_SENSITIVE,
+                ResultSet.CONCUR_UPDATABLE);
+            res = stmnt.executeQuery();
+            if (!res.next()) {
+                throw new InternalException(_loc.get("stream-exception"));
+            }
+            Clob clob = res.getClob(1);
+            Writer writer = clob.setCharacterStream(1);
+            copy(reader, writer);
+            writer.close();
+            res.updateClob(1, clob);
+            res.updateRow();
+
+        } catch (IOException ioe) {
+            throw new StoreException(ioe);
+        } finally {
+            if (res != null) 
+                try { res.close (); } catch (SQLException e) {}
+            if (stmnt != null) 
+                try { stmnt.close (); } catch (SQLException e) {}
+            if (conn != null) 
+                try { conn.close (); } catch (SQLException e) {}
+        }    
+    }
+    
+    protected long copy(InputStream in, OutputStream out) throws IOException {
+        byte[] copyBuffer = new byte[blobBufferSize];
+        long bytesCopied = 0;
+        int read = -1;
+
+        while ((read = in.read(copyBuffer, 0, copyBuffer.length)) != -1) {
+            out.write(copyBuffer, 0, read);
+            bytesCopied += read;
+        }
+        return bytesCopied;
+    }
+    
+    protected long copy(Reader reader, Writer writer) throws IOException {
+        char[] copyBuffer = new char[clobBufferSize];
+        long bytesCopied = 0;
+        int read = -1;
+
+        while ((read = reader.read(copyBuffer, 0, copyBuffer.length)) != -1) {
+            writer.write(copyBuffer, 0, read);
+            bytesCopied += read;
+        }
+
+        return bytesCopied;
+    }
+    
     /**
      * Attach CAST to the current function if necessary
      * 
@@ -3930,5 +4039,4 @@ public class DBDictionary
     public String getCastFunction(Val val, String func) {
         return func;
     }
-   
 }

openjpa-jdbc/src/main/java/org/apache/openjpa/jdbc/sql/OracleDictionary.java

@@ -49,6 +49,7 @@ import org.apache.openjpa.lib.jdbc.DelegatingDatabaseMetaData;
 import org.apache.openjpa.lib.jdbc.DelegatingPreparedStatement;
 import org.apache.openjpa.lib.util.Localizer;
 import org.apache.openjpa.util.StoreException;
+
 import serp.util.Numbers;
 
 /**
@@ -1097,4 +1098,14 @@ public class OracleDictionary
         val.appendTo(buf);
         buf.append("')");
     }
+    
+    public void insertBlobForStreamingLoad(Row row, Column col)
+        throws SQLException {
+        row.setNull(col);
+    }
+    
+    public void insertClobForStreamingLoad(Row row, Column col)
+        throws SQLException {
+        row.setNull(col);
+    }
 }

openjpa-jdbc/src/main/resources/org/apache/openjpa/jdbc/sql/localizer.properties

@@ -168,4 +168,4 @@ isolation-level-config-not-supported: This DBDictionary does not support \
 millis-query-timeout: JDBC locking does not support millisecond-granularity \
     timeouts.  Use timeouts that are multiples of 1000 for even second values.
 db-not-supported: The database product "{0}", version "{1}" is not officially supported.
-    
+stream-exception: Unexpected error recovering the row to stream the LOB.

openjpa-kernel/src/main/java/org/apache/openjpa/datacache/DataCachePCDataImpl.java

@@ -121,7 +121,7 @@ public class DataCachePCDataImpl
 
     protected Object toData(FieldMetaData fmd, Object val, StoreContext ctx) {
         // avoid caching large result set fields
-        if (fmd.isLRS())
+        if (fmd.isLRS() || fmd.isStream())
             return NULL;
         return super.toData(fmd, val, ctx);
     }

openjpa-kernel/src/main/java/org/apache/openjpa/meta/FieldMetaData.java

@@ -166,6 +166,7 @@ public class FieldMetaData
     private String[] _fgs = null;
     private String   _lfg = null;
     private Boolean _lrs = null;
+    private Boolean _stream = null;
     private String _extName = null;
     private String _factName = null;
     private String _extString = null;
@@ -1027,6 +1028,24 @@ public class FieldMetaData
         _lrs = (lrs) ? Boolean.TRUE : Boolean.FALSE;
     }
 
+    /**
+     * Whether this field is backed by a stream.
+     *
+     * @since 1.1.0
+     */
+    public boolean isStream() {
+        return _stream == Boolean.TRUE && _manage == MANAGE_PERSISTENT;
+    }
+    
+    /**
+     * Whether this field is backed by a stream.
+     *
+     * @since 1.1.0
+     */
+    public void setStream(boolean stream) {
+        _stream = (stream) ? Boolean.TRUE : Boolean.FALSE;
+    }
+    
     /**
      * Whether this field uses intermediate data when loading/storing
      * information through a {@link OpenJPAStateManager}. Defaults to true.

openjpa-kernel/src/main/java/org/apache/openjpa/meta/JavaTypes.java

@@ -18,6 +18,8 @@
  */
 package org.apache.openjpa.meta;
 
+import java.io.InputStream;
+import java.io.Reader;
 import java.io.Serializable;
 import java.lang.reflect.Array;
 import java.math.BigDecimal;
@@ -79,6 +81,8 @@ public class JavaTypes {
     public static final int PC_UNTYPED = 27;
     public static final int CALENDAR = 28;
     public static final int OID = 29;
+    public static final int INPUT_STREAM = 30;
+    public static final int INPUT_READER = 31;
 
     private static final Localizer _loc = Localizer.forPackage(JavaTypes.class);
 
@@ -156,6 +160,11 @@ public class JavaTypes {
                 return OBJECT;
             return PC_UNTYPED;
         }
+        if (type.isAssignableFrom(Reader.class))
+            return INPUT_READER;
+        if (type.isAssignableFrom (InputStream.class))
+            return INPUT_STREAM;
+            
         return OBJECT;
     }
  

openjpa-lib/src/main/java/org/apache/openjpa/lib/jdbc/DelegatingResultSet.java

@@ -436,11 +436,19 @@ public class DelegatingResultSet implements ResultSet, Closeable {
         _rs.updateBinaryStream(a, in, b);
     }
 
+    public void updateBlob(int a, Blob blob) throws SQLException {
+        _rs.updateBlob(a, blob);
+    }
+    
     public void updateCharacterStream(int a, Reader reader, int b)
         throws SQLException {
         _rs.updateCharacterStream(a, reader, b);
     }
 
+    public void updateClob(int a, Clob clob) throws SQLException {
+       _rs.updateClob(a, clob);
+    }
+    
     public void updateObject(int a, Object ob, int b) throws SQLException {
         _rs.updateObject(a, ob, b);
     }
@@ -643,18 +651,10 @@ public class DelegatingResultSet implements ResultSet, Closeable {
         throw new UnsupportedOperationException();
     }
 
-    public void updateBlob(int column, Blob blob) throws SQLException {
-        throw new UnsupportedOperationException();
-    }
-
     public void updateBlob(String columnName, Blob blob) throws SQLException {
         throw new UnsupportedOperationException();
     }
 
-    public void updateClob(int column, Clob clob) throws SQLException {
-        throw new UnsupportedOperationException();
-    }
-
     public void updateClob(String columnName, Clob clob) throws SQLException {
         throw new UnsupportedOperationException();
     }

openjpa-persistence-jdbc/src/test/java/org/apache/openjpa/jdbc/meta/strats/AbstractLobTest.java

@@ -0,0 +1,247 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.    
+ */
+
+package org.apache.openjpa.jdbc.meta.strats;
+
+import java.io.IOException;
+import javax.persistence.EntityManager;
+import javax.persistence.Query;
+
+import org.apache.openjpa.conf.OpenJPAConfiguration;
+import org.apache.openjpa.datacache.DataCachePCData;
+import org.apache.openjpa.jdbc.conf.JDBCConfiguration;
+import org.apache.openjpa.jdbc.sql.DBDictionary;
+import org.apache.openjpa.jdbc.sql.MySQLDictionary;
+import org.apache.openjpa.jdbc.sql.OracleDictionary;
+import org.apache.openjpa.jdbc.sql.SQLServerDictionary;
+import org.apache.openjpa.meta.ClassMetaData;
+import org.apache.openjpa.persistence.JPAFacadeHelper;
+import org.apache.openjpa.persistence.OpenJPAEntityManager;
+import org.apache.openjpa.persistence.test.SingleEMFTestCase;
+
+/**
+ * This abstract class defines all the tests for LOBS.
+ *
+ * @author Ignacio Andreu
+ * @since 1.1.0
+ */
+
+public abstract class AbstractLobTest extends SingleEMFTestCase {
+
+    public void setUp() throws Exception {
+        super.setUp(getLobEntityClass(), CLEAR_TABLES,
+            "openjpa.DataCache", "true",
+            "openjpa.RemoteCommitProvider", "sjvm");
+    }
+
+    public boolean isDatabaseSupported() {
+        DBDictionary dict = ((JDBCConfiguration) emf.getConfiguration())
+            .getDBDictionaryInstance();
+        if (dict instanceof MySQLDictionary ||
+            dict instanceof SQLServerDictionary ||
+            dict instanceof OracleDictionary) {
+            return true;
+        }
+        return false;
+    }
+
+    public void insert(LobEntity le) {
+        EntityManager em = emf.createEntityManager();
+        em.getTransaction().begin();
+        em.persist(le);
+        em.getTransaction().commit();
+        em.close();
+    }
+
+    public void testInsert() {
+        if (!isDatabaseSupported()) return;
+        insert(newLobEntity("oOOOOOo", 1));
+    }
+
+    public void testInsertAndSelect() throws IOException {
+        if (!isDatabaseSupported()) return;
+        String s = "oooOOOooo";
+        insert(newLobEntity(s, 1));
+        EntityManager em = emf.createEntityManager();
+        em.getTransaction().begin();
+        Query query = em.createQuery(getSelectQuery());
+        LobEntity entity = (LobEntity) query.getSingleResult();
+        assertNotNull(entity.getStream());
+        assertEquals(s, getStreamContentAsString(entity.getStream()));
+        em.getTransaction().commit();
+        em.close();
+    }
+
+    public void testInsertNull() {
+        if (!isDatabaseSupported()) return;
+        insert(newLobEntity(null, 1));
+        EntityManager em = emf.createEntityManager();
+        em.getTransaction().begin();
+        LobEntity le = (LobEntity) em.find(getLobEntityClass(), 1);
+        assertNull(le.getStream());
+        em.getTransaction().commit();
+        em.close();
+    }
+
+    public void testUpdate() throws IOException {
+        if (!isDatabaseSupported()) return;
+        insert(newLobEntity("oOOOOOo", 1));
+        EntityManager em = emf.createEntityManager();
+        em.getTransaction().begin();
+        LobEntity entity = (LobEntity) em.find(getLobEntityClass(), 1);
+        String string = "iIIIIIi";
+        changeStream(entity, string);
+        em.getTransaction().commit();
+        em.close();
+        em = emf.createEntityManager();
+        em.getTransaction().begin();
+        entity = (LobEntity) em.find(getLobEntityClass(), 1);
+        assertEquals(string, getStreamContentAsString(entity.getStream()));
+        em.getTransaction().commit();
+        em.close();
+    }
+
+    public void testDelete() {
+        if (!isDatabaseSupported()) return;
+        insert(newLobEntity("oOOOOOo", 1));
+        EntityManager em = emf.createEntityManager();
+        em.getTransaction().begin();
+        LobEntity entity = (LobEntity) em.find(getLobEntityClass(), 1);
+        em.remove(entity);
+        em.getTransaction().commit();
+        em.close();
+        em = emf.createEntityManager();
+        em.getTransaction().begin();
+        Query q = em.createQuery(getSelectQuery());
+        assertEquals(0, q.getResultList().size());
+        em.getTransaction().commit();
+        em.close();
+    }
+
+    public void testLifeCycleInsertFlushModify() {
+        if (!isDatabaseSupported()) return;
+        EntityManager em = emf.createEntityManager();
+        em.getTransaction().begin();
+        LobEntity le = newLobEntity("oOOOOOo", 1);
+        em.persist(le);
+        em.flush();
+        changeStream(le, "iIIIIIi");
+        em.getTransaction().commit();
+        em.close();
+    }
+
+    public void testLifeCycleLoadFlushModifyFlush() {
+        if (!isDatabaseSupported()) return;
+        insert(newLobEntity("oOOOOOo", 1));
+        EntityManager em = emf.createEntityManager();
+        em.getTransaction().begin();
+        LobEntity entity = (LobEntity) em.find(getLobEntityClass(), 1);
+        em.flush();
+        changeStream(entity, "iIIIIIi");
+        em.flush();
+        em.getTransaction().commit();
+        em.close();
+    }
+
+    public void testReadingMultipleTimesWithASingleConnection()
+        throws IOException {
+        if (!isDatabaseSupported()) return;
+        insert(newLobEntity("oOOOOOo", 1));
+        EntityManager em = emf.createEntityManager();
+        em.getTransaction().begin();
+        LobEntity le = (LobEntity) em.find(getLobEntityClass(), 1);
+        String string = "iIIIIIi";
+        changeStream(le, string);
+        em.getTransaction().commit();
+        em.close();
+        em = emf.createEntityManager();
+        em.getTransaction().begin();
+        le = (LobEntity) em.find(getLobEntityClass(), 1);
+        assertNotNull(le.getStream());
+        LobEntity entity = newLobEntity("oOOOOOo", 2);
+        em.persist(entity);
+        assertEquals(string, getStreamContentAsString(le.getStream()));
+        em.getTransaction().commit();
+        em.close();
+    }
+
+    public void testDataCache() {
+        if (!isDatabaseSupported()) return;
+        OpenJPAEntityManager em = emf.createEntityManager();
+
+        em.getTransaction().begin();
+        LobEntity le = newLobEntity("oOOOOOo", 1);
+        em.persist(le);
+        em.getTransaction().commit();
+        OpenJPAConfiguration conf = emf.getConfiguration();
+        Object o = em.getObjectId(le);
+        ClassMetaData meta = JPAFacadeHelper.getMetaData(le);
+        Object objectId = JPAFacadeHelper.toOpenJPAObjectId(meta, o);
+        DataCachePCData pcd =
+            conf.getDataCacheManagerInstance()
+                .getSystemDataCache().get(objectId);
+        assertFalse(pcd.isLoaded(meta.getField("stream").getIndex()));
+        em.close();
+    }
+
+    public void testSetResetAndFlush() throws IOException {
+        if (!isDatabaseSupported()) return;
+        EntityManager em = emf.createEntityManager();
+        em.getTransaction().begin();
+        LobEntity le = newLobEntity("oOOOOOo", 1);
+        em.persist(le);
+        changeStream(le, "iIIIIIi");
+        em.flush();
+        em.getTransaction().commit();
+        em.close();
+        em = emf.createEntityManager();
+        em.getTransaction().begin();
+        LobEntity entity = (LobEntity) em.find(getLobEntityClass(), 1);
+        assertEquals("iIIIIIi", getStreamContentAsString(entity.getStream()));
+        em.getTransaction().commit();
+        em.close();
+    }
+
+    public void testSetFlushAndReset() throws IOException {
+        if (!isDatabaseSupported()) return;
+        EntityManager em = emf.createEntityManager();
+        em.getTransaction().begin();
+        LobEntity le = newLobEntity("oOOOOOo", 1);
+        em.persist(le);
+        em.flush();
+        changeStream(le, "iIIIIIi");
+        LobEntity entity = (LobEntity) em.find(getLobEntityClass(), 1);
+        assertEquals("iIIIIIi", getStreamContentAsString(entity.getStream()));
+        em.getTransaction().commit();
+        em.close();
+    }
+
+    protected abstract Class getLobEntityClass();
+
+    protected abstract String getStreamContentAsString(Object o)
+        throws IOException;
+
+    protected abstract LobEntity newLobEntity(String s, int id);
+
+    protected abstract LobEntity newLobEntityForLoadContent(String s, int id);
+
+    protected abstract String getSelectQuery();
+
+    protected abstract void changeStream(LobEntity le, String s);
+}

openjpa-persistence-jdbc/src/test/java/org/apache/openjpa/jdbc/meta/strats/InputStreamLobEntity.java

@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.    
+ */
+package org.apache.openjpa.jdbc.meta.strats;
+
+import java.io.InputStream;
+import javax.persistence.Entity;
+import javax.persistence.Id;
+
+import org.apache.openjpa.persistence.Persistent;
+
+/**
+ * An entity with an InputStream.
+ *
+ * @author Ignacio Andreu
+ * @since 1.1.0
+ */
+
+@Entity
+public class InputStreamLobEntity implements LobEntity {
+
+    @Id
+    private int id;
+
+    @Persistent
+    private InputStream stream;
+
+    public int getId() {
+        return id;
+    }
+
+    public void setId(int id) {
+        this.id = id;
+    }
+
+    public Object getStream() {
+        return stream;
+    }
+
+    public void setStream(Object o) {
+        stream = (InputStream) o;
+    }
+}

openjpa-persistence-jdbc/src/test/java/org/apache/openjpa/jdbc/meta/strats/InputStreamLobTest.java

@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.    
+ */
+package org.apache.openjpa.jdbc.meta.strats;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * Defines all the abstract methods from AbstractLobTest to tests the
+ * the LOB support with an InputStream.
+ *
+ * @author Ignacio Andreu
+ * @since 1.1.0
+ */
+
+public class InputStreamLobTest extends AbstractLobTest {
+
+    protected LobEntity newLobEntity(String s, int id) {
+        InputStreamLobEntity isle = new InputStreamLobEntity();
+        isle.setId(id);
+        if (s != null) {
+            isle.setStream(new ByteArrayInputStream(s.getBytes()));
+        } else {
+            isle.setStream(null);
+        }
+        return isle;
+    }
+
+    protected LobEntity newLobEntityForLoadContent(String s, int id) {
+        InputStreamLobEntity isle = new InputStreamLobEntity();
+        isle.setId(id);
+        isle.setStream(new InputStreamWrapper(s));
+        return isle;
+    }
+
+    protected Class getLobEntityClass() {
+        return InputStreamLobEntity.class;
+    }
+
+    protected String getSelectQuery() {
+        return "SELECT o FROM InputStreamLobEntity o";
+    }
+
+    protected String getStreamContentAsString(Object o) throws IOException {
+        InputStream is = (InputStream) o;
+        String content = "";
+        byte[] bs = new byte[4];
+        int read = -1;
+        do {
+            read = is.read(bs);
+            if (read == -1) {
+                return content;
+            }
+            content = content + (new String(bs)).substring(0, read);
+        } while (true);
+    }
+
+    protected void changeStream(LobEntity le, String s) {
+        le.setStream(new ByteArrayInputStream(s.getBytes()));
+    }
+}

openjpa-persistence-jdbc/src/test/java/org/apache/openjpa/jdbc/meta/strats/InputStreamWrapper.java

@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.    
+ */
+package org.apache.openjpa.jdbc.meta.strats;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * This class is used to kwon where the content of the InputStream is load.
+ * If the content is load out of the flush then throws a
+ * UnsupportedOperationException
+ *
+ * @author Ignacio Andreu
+ * @since 1.1.0
+ */
+
+public class InputStreamWrapper extends InputStream {
+
+    private InputStream is;
+
+    public InputStreamWrapper(String s) {
+        this.is = new ByteArrayInputStream(s.getBytes());
+    }
+
+    public int read() throws IOException {
+        throw new UnsupportedOperationException();
+    }
+
+    public int available() throws IOException {
+        return is.available();
+    }
+
+    public void close() throws IOException {
+        is.close();
+    }
+
+    public int read(byte[] b, int off, int len) throws IOException {
+        StackTraceElement[] ste = Thread.currentThread().getStackTrace();
+        for (StackTraceElement element : ste) {
+            if ("flush".equals(element.getMethodName())) {
+                return is.read(b, off, len);
+            }
+        }
+        throw new UnsupportedOperationException();
+    }
+
+    public int read(byte[] b) throws IOException {
+        throw new UnsupportedOperationException();
+    }
+}

openjpa-persistence-jdbc/src/test/java/org/apache/openjpa/jdbc/meta/strats/LobEntity.java

@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.    
+ */
+
+package org.apache.openjpa.jdbc.meta.strats;
+
+/**
+ * Defines the methods for the entities
+ *
+ * @author Ignacio Andreu
+ * @since 1.1.0
+ */
+public interface LobEntity {
+
+    public void setStream(Object o);
+
+    public void setId(int id);
+
+    public Object getStream();
+
+    public int getId();
+}

openjpa-persistence-jdbc/src/test/java/org/apache/openjpa/jdbc/meta/strats/ReaderLobEntity.java

@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.    
+ */
+package org.apache.openjpa.jdbc.meta.strats;
+
+import java.io.Reader;
+import javax.persistence.Entity;
+import javax.persistence.Id;
+
+import org.apache.openjpa.persistence.Persistent;
+
+/**
+ * An entity with a Reader field.
+ *
+ * @author Ignacio Andreu
+ * @since 1.1.0
+ */
+
+@Entity
+public class ReaderLobEntity implements LobEntity {
+
+    @Id
+    int id;
+
+    @Persistent
+    Reader stream;
+
+    public int getId() {
+        return id;
+    }
+
+    public void setId(int id) {
+        this.id = id;
+    }
+
+    public Object getStream() {
+        return stream;
+    }
+
+    public void setStream(Object o) {
+        this.stream = (Reader) o;
+    }
+}

openjpa-persistence-jdbc/src/test/java/org/apache/openjpa/jdbc/meta/strats/ReaderLobTest.java

@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.    
+ */
+package org.apache.openjpa.jdbc.meta.strats;
+
+import java.io.CharArrayReader;
+import java.io.IOException;
+import java.io.Reader;
+
+/**
+ * Defines all the abstract methods from AbstractLobTest to tests the
+ * the LOB support with a Reader.
+ *
+ * @author Ignacio Andreu
+ * @since 1.1.0
+ */
+
+public class ReaderLobTest extends AbstractLobTest {
+
+    protected LobEntity newLobEntity(String s, int id) {
+        ReaderLobEntity rle = new ReaderLobEntity();
+        rle.setId(id);
+        if (s != null) {
+            rle.setStream(new CharArrayReader(s.toCharArray()));
+        } else {
+            rle.setStream(null);
+        }
+        return rle;
+    }
+
+    protected LobEntity newLobEntityForLoadContent(String s, int id) {
+        ReaderLobEntity rle = new ReaderLobEntity();
+        rle.setId(id);
+        rle.setStream(new ReaderWrapper(s));
+        return rle;
+    }
+
+    protected Class getLobEntityClass() {
+        return ReaderLobEntity.class;
+    }
+
+    protected String getSelectQuery() {
+        return "SELECT o FROM ReaderLobEntity o";
+    }
+
+    protected String getStreamContentAsString(Object o) throws IOException {
+        Reader r = (Reader) o;
+        String content = "";
+        char[] cs = new char[4];
+        int read = -1;
+        do {
+            read = r.read(cs);
+            if (read == -1) {
+                return content;
+            }
+            content = content + (new String(cs)).substring(0, read);
+        } while (true);
+    }
+
+    protected void changeStream(LobEntity le, String s) {
+        le.setStream(new CharArrayReader(s.toCharArray()));
+    }
+}

openjpa-persistence-jdbc/src/test/java/org/apache/openjpa/jdbc/meta/strats/ReaderWrapper.java

@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.    
+ */
+package org.apache.openjpa.jdbc.meta.strats;
+
+import java.io.CharArrayReader;
+import java.io.IOException;
+import java.io.Reader;
+
+/**
+ * This class is used to kwon where the content of the Reader is load.
+ * If the content is load out of the flush then throws a
+ * UnsupportedOperationException
+ *
+ * @author Ignacio Andreu
+ * @since 1.1.0
+ */
+
+public class ReaderWrapper extends Reader {
+
+    private Reader reader;
+
+    public ReaderWrapper(String s) {
+        this.reader = new CharArrayReader(s.toCharArray());
+    }
+
+    public void close() throws IOException {
+        reader.close();
+    }
+
+    public int read(char[] cbuf, int off, int len) throws IOException {
+        StackTraceElement[] ste = Thread.currentThread().getStackTrace();
+        for (StackTraceElement element : ste) {
+            if ("flush".equals(element.getMethodName())) {
+                return reader.read(cbuf, off, len);
+            }
+        }
+        throw new UnsupportedOperationException();
+    }
+}