Author: shawkins Date: 2012-02-29 17:59:27 -0500 (Wed, 29 Feb 2012) New Revision: 3905 Modified: branches/7.7.x/api/src/main/java/org/teiid/translator/ExecutionFactory.java branches/7.7.x/build/kits/jboss-container/teiid-releasenotes.html branches/7.7.x/connectors/translator-jdbc/src/main/java/org/teiid/translator/jdbc/JDBCBaseExecution.java branches/7.7.x/engine/src/main/java/org/teiid/common/buffer/BufferManager.java branches/7.7.x/engine/src/main/java/org/teiid/common/buffer/LobManager.java branches/7.7.x/engine/src/main/java/org/teiid/common/buffer/impl/BufferManagerImpl.java branches/7.7.x/engine/src/main/java/org/teiid/dqp/internal/datamgr/ConnectorWork.java branches/7.7.x/engine/src/main/java/org/teiid/dqp/internal/datamgr/ConnectorWorkItem.java branches/7.7.x/engine/src/main/java/org/teiid/dqp/internal/process/DataTierTupleSource.java branches/7.7.x/engine/src/test/java/org/teiid/dqp/internal/process/TestDataTierManager.java branches/7.7.x/engine/src/test/java/org/teiid/dqp/service/AutoGenDataService.java Log: TEIID-1937 adding support for lob value copying by the engine. Modified: branches/7.7.x/api/src/main/java/org/teiid/translator/ExecutionFactory.java =================================================================== --- branches/7.7.x/api/src/main/java/org/teiid/translator/ExecutionFactory.java 2012-02-29 16:54:17 UTC (rev 3904) +++ branches/7.7.x/api/src/main/java/org/teiid/translator/ExecutionFactory.java 2012-02-29 22:59:27 UTC (rev 3905) @@ -108,6 +108,7 @@ private boolean requiresCriteria; private int maxInSize = DEFAULT_MAX_IN_CRITERIA_SIZE; private int maxDependentInPredicates = DEFAULT_MAX_IN_CRITERIA_SIZE; + private boolean copyLobs; private LinkedList<FunctionMethod> pushdownFunctionMethods = new LinkedList<FunctionMethod>(); @@ -129,8 +130,17 @@ public void setImmutable(boolean arg0) { this.immutable = arg0; - } + } + + @TranslatorProperty(display="Copy LOBs",description="If true, returned LOBs will be copied, rather than streamed from the source",advanced=true) + public boolean isCopyLobs() { + return copyLobs; + } + public void setCopyLobs(boolean copyLobs) { + this.copyLobs = copyLobs; + } + /** * Return a connection object from the given connection factory. * Modified: branches/7.7.x/build/kits/jboss-container/teiid-releasenotes.html =================================================================== --- branches/7.7.x/build/kits/jboss-container/teiid-releasenotes.html 2012-02-29 16:54:17 UTC (rev 3904) +++ branches/7.7.x/build/kits/jboss-container/teiid-releasenotes.html 2012-02-29 22:59:27 UTC (rev 3905) @@ -30,6 +30,7 @@ <LI><B>Salesforce Aggregates</B> - salesforce pushdown queries now support GROUP BY, HAVING, and the standard aggregate functions. <LI><B>Comparable LOBs</B> - the system property org.teiid.comparableLobs can be set to use CLOB and BLOB values in comparison/sorting/grouping operations. <LI><B>Padded String Comparison</B> - the system property org.teiid.padSpace can be set to effectively right pad strings to the same length for comparison. + <LI><B>Copy LOBs</B> - added the copyLobs property to indicate that lob values should be copied by the engine rather than being held by reference. </UL> <h2><a name="Compatibility">Compatibility Issues</a></h2> Modified: branches/7.7.x/connectors/translator-jdbc/src/main/java/org/teiid/translator/jdbc/JDBCBaseExecution.java =================================================================== --- branches/7.7.x/connectors/translator-jdbc/src/main/java/org/teiid/translator/jdbc/JDBCBaseExecution.java 2012-02-29 16:54:17 UTC (rev 3904) +++ branches/7.7.x/connectors/translator-jdbc/src/main/java/org/teiid/translator/jdbc/JDBCBaseExecution.java 2012-02-29 22:59:27 UTC (rev 3905) @@ -35,9 +35,9 @@ import org.teiid.logging.LogConstants; import org.teiid.logging.LogManager; import org.teiid.logging.MessageLevel; -import org.teiid.translator.TranslatorException; import org.teiid.translator.Execution; import org.teiid.translator.ExecutionContext; +import org.teiid.translator.TranslatorException; /** Modified: branches/7.7.x/engine/src/main/java/org/teiid/common/buffer/BufferManager.java =================================================================== --- branches/7.7.x/engine/src/main/java/org/teiid/common/buffer/BufferManager.java 2012-02-29 16:54:17 UTC (rev 3904) +++ branches/7.7.x/engine/src/main/java/org/teiid/common/buffer/BufferManager.java 2012-02-29 22:59:27 UTC (rev 3905) @@ -25,6 +25,7 @@ import java.util.List; import org.teiid.core.TeiidComponentException; +import org.teiid.core.types.Streamable; import org.teiid.query.sql.symbol.Expression; @@ -138,5 +139,8 @@ * @param additional * @return */ - int reserveAdditionalBuffers(int additional); + int reserveAdditionalBuffers(int additional); + + Streamable<?> persistLob(final Streamable<?> lob, + final FileStore store, byte[] bytes) throws TeiidComponentException; } Modified: branches/7.7.x/engine/src/main/java/org/teiid/common/buffer/LobManager.java =================================================================== --- branches/7.7.x/engine/src/main/java/org/teiid/common/buffer/LobManager.java 2012-02-29 16:54:17 UTC (rev 3904) +++ branches/7.7.x/engine/src/main/java/org/teiid/common/buffer/LobManager.java 2012-02-29 22:59:27 UTC (rev 3905) @@ -35,8 +35,9 @@ import java.util.List; import java.util.Map; +import javax.sql.rowset.serial.SerialBlob; + import org.teiid.core.TeiidComponentException; -import org.teiid.core.types.BaseLob; import org.teiid.core.types.BlobImpl; import org.teiid.core.types.BlobType; import org.teiid.core.types.ClobImpl; @@ -190,60 +191,79 @@ byte[] bytes = new byte[1 << 14]; for (Map.Entry<String, LobHolder> entry : this.lobReferences.entrySet()) { - entry.getValue().lob = persistLob(entry.getValue().lob, lobStore, bytes); + entry.getValue().lob = detachLob(entry.getValue().lob, lobStore, bytes); } } - private Streamable<?> persistLob(final Streamable<?> lob, final FileStore store, byte[] bytes) throws TeiidComponentException { + public Streamable<?> detachLob(final Streamable<?> lob, final FileStore store, byte[] bytes) throws TeiidComponentException { + // if this is not attached, just return + if (InputStreamFactory.getStorageMode(lob) != StorageMode.OTHER) { + return lob; + } - // if this is already saved to disk just return - if (lob.getReference() instanceof BaseLob) { - try { - BaseLob baseLob = (BaseLob)lob.getReference(); - InputStreamFactory isf = baseLob.getStreamFactory(); - if (isf.getStorageMode() == StorageMode.PERSISTENT) { + return persistLob(lob, store, bytes, inlineLobs, maxMemoryBytes); + } + + public static Streamable<?> persistLob(final Streamable<?> lob, + final FileStore store, byte[] bytes, boolean inlineLobs, int maxMemoryBytes) throws TeiidComponentException { + try { + //inline + long byteLength = lob.length()*(lob instanceof ClobType?2:1); + if (lob.getReferenceStreamId() == null || (inlineLobs + && (byteLength <= maxMemoryBytes))) { + lob.setReferenceStreamId(null); + if (InputStreamFactory.getStorageMode(lob) == StorageMode.MEMORY) { return lob; } - } catch (SQLException e) { - // go through regular persistence. + + if (lob instanceof BlobType) { + BlobType b = (BlobType)lob; + byte[] blobBytes = b.getBytes(1, (int)byteLength); + b.setReference(new SerialBlob(blobBytes)); + return b; + } else if (lob instanceof ClobType) { + ClobType c = (ClobType)lob; + String s = c.getSubString(1, (int)(byteLength>>>1)); + c.setReference(ClobImpl.createClob(s.toCharArray())); + return c; + } else { + XMLType x = (XMLType)lob; + String s = x.getString(); + x.setReference(new SQLXMLImpl(s)); + return x; + } } - } - long offset = store.getLength(); - int length = 0; - Streamable<?> persistedLob; - - try { + InputStream is = null; if (lob instanceof BlobType) { is = new BlobInputStreamFactory((Blob)lob).getInputStream(); - } - else if (lob instanceof ClobType) { + } else if (lob instanceof ClobType) { is = new ClobInputStreamFactory((Clob)lob).getInputStream(); } else { is = new SQLXMLInputStreamFactory((SQLXML)lob).getInputStream(); } + + long offset = store.getLength(); + Streamable<?> persistedLob; + OutputStream fsos = store.createOutputStream(); - length = ObjectConverterUtil.write(fsos, is, bytes, -1); - } catch (IOException e) { - throw new TeiidComponentException(e); - } - - // re-construct the new lobs based on the file store - final long lobOffset = offset; - final int lobLength = length; - InputStreamFactory isf = new InputStreamFactory() { - @Override - public InputStream getInputStream() throws IOException { - return store.createInputStream(lobOffset, lobLength); - } + byteLength = ObjectConverterUtil.write(fsos, is, bytes, -1); - @Override - public StorageMode getStorageMode() { - return StorageMode.PERSISTENT; - } - }; + // re-construct the new lobs based on the file store + final long lobOffset = offset; + final long lobLength = byteLength; + InputStreamFactory isf = new InputStreamFactory() { + @Override + public InputStream getInputStream() throws IOException { + return store.createInputStream(lobOffset, lobLength); + } + + @Override + public StorageMode getStorageMode() { + return StorageMode.PERSISTENT; + } + }; - try { if (lob instanceof BlobType) { persistedLob = new BlobType(new BlobImpl(isf)); } @@ -255,10 +275,12 @@ ((XMLType)persistedLob).setEncoding(((XMLType)lob).getEncoding()); ((XMLType)persistedLob).setType(((XMLType)lob).getType()); } + return persistedLob; } catch (SQLException e) { throw new TeiidComponentException(e); - } - return persistedLob; + } catch (IOException e) { + throw new TeiidComponentException(e); + } } public int getLobCount() { @@ -267,6 +289,5 @@ public void remove() { this.lobReferences.clear(); - } } Modified: branches/7.7.x/engine/src/main/java/org/teiid/common/buffer/impl/BufferManagerImpl.java =================================================================== --- branches/7.7.x/engine/src/main/java/org/teiid/common/buffer/impl/BufferManagerImpl.java 2012-02-29 16:54:17 UTC (rev 3904) +++ branches/7.7.x/engine/src/main/java/org/teiid/common/buffer/impl/BufferManagerImpl.java 2012-02-29 22:59:27 UTC (rev 3905) @@ -48,23 +48,13 @@ import org.teiid.client.BatchSerializer; import org.teiid.client.ResizingArrayList; -import org.teiid.common.buffer.AutoCleanupUtil; -import org.teiid.common.buffer.BatchManager; -import org.teiid.common.buffer.BufferManager; -import org.teiid.common.buffer.Cache; -import org.teiid.common.buffer.CacheEntry; -import org.teiid.common.buffer.CacheKey; -import org.teiid.common.buffer.FileStore; -import org.teiid.common.buffer.LobManager; -import org.teiid.common.buffer.STree; -import org.teiid.common.buffer.Serializer; -import org.teiid.common.buffer.StorageManager; -import org.teiid.common.buffer.TupleBuffer; +import org.teiid.common.buffer.*; import org.teiid.common.buffer.AutoCleanupUtil.Removable; import org.teiid.common.buffer.LobManager.ReferenceMode; import org.teiid.core.TeiidComponentException; import org.teiid.core.TeiidRuntimeException; import org.teiid.core.types.DataTypeManager; +import org.teiid.core.types.Streamable; import org.teiid.core.types.DataTypeManager.WeakReferenceHashedValueCache; import org.teiid.dqp.internal.process.DQPConfiguration; import org.teiid.logging.LogConstants; @@ -282,7 +272,7 @@ } long count = readCount.incrementAndGet(); if (LogManager.isMessageToBeRecorded(LogConstants.CTX_BUFFER_MGR, MessageLevel.DETAIL)) { - LogManager.logDetail(LogConstants.CTX_BUFFER_MGR, id, id, "reading batch", batch, "from storage, total reads:", count); //$NON-NLS-1$ //$NON-NLS-2$ + LogManager.logDetail(LogConstants.CTX_BUFFER_MGR, id, "reading batch", batch, "from storage, total reads:", count); //$NON-NLS-1$ //$NON-NLS-2$ } ce = cache.get(o, batch, this.ref); if (ce == null) { @@ -993,4 +983,10 @@ return memoryEntries.size(); } + @Override + public Streamable<?> persistLob(Streamable<?> lob, FileStore store, + byte[] bytes) throws TeiidComponentException { + return LobManager.persistLob(lob, store, bytes, inlineLobs, DataTypeManager.MAX_LOB_MEMORY_BYTES); + } + } Modified: branches/7.7.x/engine/src/main/java/org/teiid/dqp/internal/datamgr/ConnectorWork.java =================================================================== --- branches/7.7.x/engine/src/main/java/org/teiid/dqp/internal/datamgr/ConnectorWork.java 2012-02-29 16:54:17 UTC (rev 3904) +++ branches/7.7.x/engine/src/main/java/org/teiid/dqp/internal/datamgr/ConnectorWork.java 2012-02-29 22:59:27 UTC (rev 3905) @@ -45,4 +45,6 @@ boolean isDataAvailable(); + boolean copyLobs(); + } \ No newline at end of file Modified: branches/7.7.x/engine/src/main/java/org/teiid/dqp/internal/datamgr/ConnectorWorkItem.java =================================================================== --- branches/7.7.x/engine/src/main/java/org/teiid/dqp/internal/datamgr/ConnectorWorkItem.java 2012-02-29 16:54:17 UTC (rev 3904) +++ branches/7.7.x/engine/src/main/java/org/teiid/dqp/internal/datamgr/ConnectorWorkItem.java 2012-02-29 22:59:27 UTC (rev 3905) @@ -371,5 +371,10 @@ public boolean isDataAvailable() { return this.securityContext.isDataAvailable(); } + + @Override + public boolean copyLobs() { + return this.connector.isCopyLobs(); + } } \ No newline at end of file Modified: branches/7.7.x/engine/src/main/java/org/teiid/dqp/internal/process/DataTierTupleSource.java =================================================================== --- branches/7.7.x/engine/src/main/java/org/teiid/dqp/internal/process/DataTierTupleSource.java 2012-02-29 16:54:17 UTC (rev 3904) +++ branches/7.7.x/engine/src/main/java/org/teiid/dqp/internal/process/DataTierTupleSource.java 2012-02-29 22:59:27 UTC (rev 3905) @@ -87,6 +87,9 @@ private boolean[] convertToRuntimeType; private boolean[] convertToDesiredRuntimeType; + private boolean[] isLob; + private FileStore lobStore; + private byte[] lobBuffer; private Class<?>[] schema; private int limit = -1; @@ -116,11 +119,13 @@ this.schema = new Class[symbols.size()]; this.convertToDesiredRuntimeType = new boolean[symbols.size()]; this.convertToRuntimeType = new boolean[symbols.size()]; + this.isLob = new boolean[symbols.size()]; for (int i = 0; i < symbols.size(); i++) { SingleElementSymbol symbol = symbols.get(i); this.schema[i] = symbol.getType(); this.convertToDesiredRuntimeType[i] = true; this.convertToRuntimeType[i] = true; + this.isLob[i] = DataTypeManager.isLOB(this.schema[i]); } Assertion.isNull(workItem.getConnectorRequest(aqr.getAtomicRequestID())); @@ -140,7 +145,7 @@ }, this, 100); } - private List<?> correctTypes(List<Object> row) throws TransformationException { + private List<?> correctTypes(List<Object> row) throws TransformationException, TeiidComponentException { //TODO: add a proper intermediate schema for (int i = 0; i < row.size(); i++) { Object value = row.get(i); @@ -148,12 +153,11 @@ continue; } if (convertToRuntimeType[i]) { - boolean lob = !arm.supportsCloseWithLobs() && DataTypeManager.isLOB(value.getClass()); Object result = convertToRuntimeType(value, this.schema[i]); if (value == result && !DataTypeManager.DefaultDataClasses.OBJECT.equals(this.schema[i])) { convertToRuntimeType[i] = false; } else { - if (lob && DataTypeManager.isLOB(result.getClass()) && DataTypeManager.isLOB(this.schema[i])) { + if (isLob[i] && !cwi.copyLobs() && !arm.supportsCloseWithLobs() && DataTypeManager.isLOB(value.getClass())) { explicitClose = true; } row.set(i, result); @@ -163,13 +167,19 @@ if (convertToDesiredRuntimeType[i]) { if (value != null) { Object result = DataTypeManager.transformValue(value, value.getClass(), this.schema[i]); - if (value == result) { + if (isLob[i] && cwi.copyLobs()) { + if (lobStore == null) { + lobStore = dtm.getBufferManager().createFileStore("lobs"); //$NON-NLS-1$ + lobBuffer = new byte[1 << 14]; + } + result = dtm.getBufferManager().persistLob((Streamable<?>) result, lobStore, lobBuffer); + } else if (value == result) { convertToDesiredRuntimeType[i] = false; continue; } row.set(i, result); } - } else { + } else if (DataTypeManager.isValueCacheEnabled()) { row.set(i, DataTypeManager.getCanonicalValue(value)); } } @@ -393,6 +403,8 @@ * @see TupleSource#closeSource() */ public void closeSource() { + lobBuffer = null; + lobStore = null; //can still be referenced by lobs and will be cleaned-up by reference cancelAsynch = true; if (!explicitClose) { fullyCloseSource(); Modified: branches/7.7.x/engine/src/test/java/org/teiid/dqp/internal/process/TestDataTierManager.java =================================================================== --- branches/7.7.x/engine/src/test/java/org/teiid/dqp/internal/process/TestDataTierManager.java 2012-02-29 16:54:17 UTC (rev 3904) +++ branches/7.7.x/engine/src/test/java/org/teiid/dqp/internal/process/TestDataTierManager.java 2012-02-29 22:59:27 UTC (rev 3905) @@ -24,6 +24,8 @@ import static org.junit.Assert.*; +import java.util.List; + import org.junit.Before; import org.junit.Test; import org.mockito.Mockito; @@ -31,6 +33,9 @@ import org.teiid.client.RequestMessage; import org.teiid.client.SourceWarning; import org.teiid.common.buffer.BlockedException; +import org.teiid.core.types.ClobType; +import org.teiid.core.types.InputStreamFactory; +import org.teiid.core.types.InputStreamFactory.StorageMode; import org.teiid.dqp.internal.datamgr.ConnectorManagerRepository; import org.teiid.dqp.internal.datamgr.FakeTransactionService; import org.teiid.dqp.message.AtomicRequestMessage; @@ -45,6 +50,7 @@ import org.teiid.query.unittest.RealMetadataFactory; import org.teiid.query.util.CommandContext; +@SuppressWarnings("nls") public class TestDataTierManager { private DQPCore rm; @@ -109,6 +115,33 @@ info = new DataTierTupleSource(request, workItem, connectorManager.registerRequest(request), dtm, limit); } + @Test public void testCopyLobs() throws Exception { + connectorManager.copyLobs = true; + helpSetup("SELECT cast(stringkey as clob) from bqt1.smalla", 1); + for (int i = 0; i < 10;) { + try { + List<?> tuple = info.nextTuple(); + ClobType clob = (ClobType)tuple.get(0); + assertEquals(StorageMode.MEMORY, InputStreamFactory.getStorageMode(clob)); + i++; + } catch (BlockedException e) { + Thread.sleep(50); + } + } + connectorManager.copyLobs = false; + helpSetup("SELECT cast(stringkey as clob) from bqt1.smalla", 1); + for (int i = 0; i < 10;) { + try { + List<?> tuple = info.nextTuple(); + ClobType clob = (ClobType)tuple.get(0); + assertEquals(StorageMode.OTHER, InputStreamFactory.getStorageMode(clob)); + i++; + } catch (BlockedException e) { + Thread.sleep(50); + } + } + } + @Test public void testDataTierTupleSource() throws Exception { helpSetup(1); for (int i = 0; i < 10;) { Modified: branches/7.7.x/engine/src/test/java/org/teiid/dqp/service/AutoGenDataService.java =================================================================== --- branches/7.7.x/engine/src/test/java/org/teiid/dqp/service/AutoGenDataService.java 2012-02-29 16:54:17 UTC (rev 3904) +++ branches/7.7.x/engine/src/test/java/org/teiid/dqp/service/AutoGenDataService.java 2012-02-29 22:59:27 UTC (rev 3905) @@ -22,8 +22,12 @@ package org.teiid.dqp.service; +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; import java.math.BigDecimal; import java.math.BigInteger; +import java.sql.Clob; import java.util.ArrayList; import java.util.Arrays; import java.util.Iterator; @@ -31,7 +35,10 @@ import java.util.concurrent.atomic.AtomicInteger; import org.teiid.core.TeiidComponentException; +import org.teiid.core.types.ClobImpl; import org.teiid.core.types.DataTypeManager; +import org.teiid.core.types.InputStreamFactory; +import org.teiid.core.types.Streamable; import org.teiid.dqp.internal.datamgr.ConnectorManager; import org.teiid.dqp.internal.datamgr.ConnectorWork; import org.teiid.dqp.internal.datamgr.ConnectorWorkItem; @@ -50,6 +57,7 @@ * This data service will automatically generate results when called with a query - basically * the same as the old loopback connector. */ +@SuppressWarnings("nls") public class AutoGenDataService extends ConnectorManager{ // Number of rows that will be generated for each query @@ -62,6 +70,7 @@ private final AtomicInteger closeCount = new AtomicInteger(); private boolean useIntCounter; public boolean addWarning; + public boolean copyLobs; public AutoGenDataService() { super("FakeConnector","FakeConnector"); //$NON-NLS-1$ //$NON-NLS-2$ @@ -166,6 +175,11 @@ } + @Override + public boolean copyLobs() { + return copyLobs; + } + }; } @@ -202,6 +216,12 @@ private static final Double DOUBLE_VAL = new Double(0.0); private static final Character CHAR_VAL = new Character('c'); private static final Byte BYTE_VAL = new Byte((byte)0); + private static final Clob CLOB_VAL = new ClobImpl(new InputStreamFactory() { + @Override + public InputStream getInputStream() throws IOException { + return new ByteArrayInputStream("hello world".getBytes(Streamable.CHARSET)); + } + }, -1); private static final Boolean BOOLEAN_VAL = Boolean.FALSE; private static final BigInteger BIG_INTEGER_VAL = new BigInteger("0"); //$NON-NLS-1$ private static final BigDecimal BIG_DECIMAL_VAL = new BigDecimal("0"); //$NON-NLS-1$ @@ -238,6 +258,8 @@ return TIME_VAL; } else if(type.equals(DataTypeManager.DefaultDataClasses.TIMESTAMP)) { return TIMESTAMP_VAL; + } else if(type.equals(DataTypeManager.DefaultDataClasses.CLOB)) { + return CLOB_VAL; } else { return null; }