7:47 p.m.
Author: shawkins
Date: 2010-12-24 13:47:11 -0500 (Fri, 24 Dec 2010)
New Revision: 2803
Modified:
trunk/common-core/src/main/java/org/teiid/core/types/InputStreamFactory.java
trunk/common-core/src/main/java/org/teiid/core/util/ObjectConverterUtil.java
trunk/engine/src/main/java/org/teiid/common/buffer/LobManager.java
trunk/engine/src/main/java/org/teiid/common/buffer/TupleBuffer.java
trunk/engine/src/main/java/org/teiid/common/buffer/impl/BufferManagerImpl.java
trunk/engine/src/main/java/org/teiid/dqp/internal/process/RequestWorkItem.java
trunk/engine/src/main/java/org/teiid/query/optimizer/relational/RelationalPlanner.java
trunk/engine/src/main/java/org/teiid/query/processor/xml/XMLUtil.java
Log:
TEIID-1227 expanding the persistent check and ensuring that procedures with lob params are
not cached.
Modified: trunk/common-core/src/main/java/org/teiid/core/types/InputStreamFactory.java
===================================================================
---
trunk/common-core/src/main/java/org/teiid/core/types/InputStreamFactory.java 2010-12-24
02:09:29 UTC (rev 2802)
+++
trunk/common-core/src/main/java/org/teiid/core/types/InputStreamFactory.java 2010-12-24
18:47:11 UTC (rev 2803)
@@ -83,6 +83,10 @@
return null;
}
+ public boolean isPersistent() {
+ return false;
+ }
+
public static class FileInputStreamFactory extends InputStreamFactory {
private File f;
@@ -102,6 +106,11 @@
return new BufferedInputStream(new FileInputStream(f));
}
+ @Override
+ public boolean isPersistent() {
+ return true;
+ }
+
}
public static class ClobInputStreamFactory extends InputStreamFactory implements
DataSource {
Modified: trunk/common-core/src/main/java/org/teiid/core/util/ObjectConverterUtil.java
===================================================================
---
trunk/common-core/src/main/java/org/teiid/core/util/ObjectConverterUtil.java 2010-12-24
02:09:29 UTC (rev 2802)
+++
trunk/common-core/src/main/java/org/teiid/core/util/ObjectConverterUtil.java 2010-12-24
18:47:11 UTC (rev 2803)
@@ -95,10 +95,9 @@
return out.toByteArray();
}
- public static void write(final OutputStream out, final InputStream is, int length)
throws IOException {
+ public static void write(final OutputStream out, final InputStream is, byte[]
l_buffer, int length) throws IOException {
int writen = 0;
try {
- byte[] l_buffer = new byte[DEFAULT_READING_SIZE]; // buffer holding bytes to be
transferred
int l_nbytes = 0; // Number of bytes read
while ((l_nbytes = is.read(l_buffer)) != -1) {
if (length != -1 && writen > length - l_nbytes) {
@@ -117,6 +116,10 @@
}
}
+ public static void write(final OutputStream out, final InputStream is, int length)
throws IOException {
+ write(out, is, new byte[DEFAULT_READING_SIZE], length); // buffer holding bytes to
be transferred
+ }
+
public static void write(final Writer out, final Reader is, int length) throws
IOException {
int writen = 0;
try {
Modified: trunk/engine/src/main/java/org/teiid/common/buffer/LobManager.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/common/buffer/LobManager.java 2010-12-24 02:09:29
UTC (rev 2802)
+++ trunk/engine/src/main/java/org/teiid/common/buffer/LobManager.java 2010-12-24 18:47:11
UTC (rev 2803)
@@ -22,7 +22,6 @@
package org.teiid.common.buffer;
-import java.io.BufferedOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
@@ -30,14 +29,13 @@
import java.sql.Clob;
import java.sql.SQLException;
import java.sql.SQLXML;
-import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
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;
@@ -47,8 +45,8 @@
import org.teiid.core.types.SQLXMLImpl;
import org.teiid.core.types.Streamable;
import org.teiid.core.types.XMLType;
+import org.teiid.core.util.ObjectConverterUtil;
import org.teiid.query.QueryPlugin;
-import org.teiid.query.processor.xml.XMLUtil.FileStoreInputStreamFactory;
import org.teiid.query.sql.symbol.Expression;
/**
@@ -56,9 +54,7 @@
* TODO: for temp tables we may need to have a copy by value management strategy
*/
public class LobManager {
- private static final int IO_BUFFER_SIZE = 1 << 14;
private Map<String, Streamable<?>> lobReferences = new
ConcurrentHashMap<String, Streamable<?>>();
- private Map<String, Streamable<?>> lobFilestores = new
ConcurrentHashMap<String, Streamable<?>>();
public void updateReferences(int[] lobIndexes, List<?> tuple)
throws TeiidComponentException {
@@ -78,26 +74,13 @@
}
public Streamable<?> getLobReference(String id) throws TeiidComponentException
{
- Streamable<?> lob = null;
- if (this.lobReferences != null) {
- lob = this.lobReferences.get(id);
- }
-
+ Streamable<?> lob = this.lobReferences.get(id);
if (lob == null) {
- lob = this.lobFilestores.get(id);
- }
-
- if (lob == null) {
throw new
TeiidComponentException(QueryPlugin.Util.getString("ProcessWorker.wrongdata"));
//$NON-NLS-1$
}
return lob;
}
- public void clear() {
- this.lobReferences.clear();
- this.lobFilestores.clear();
- }
-
public static int[] getLobIndexes(List expressions) {
if (expressions == null) {
return null;
@@ -116,54 +99,33 @@
return Arrays.copyOf(result, resultIndex);
}
- public Collection<Streamable<?>> getLobReferences(){
- return lobReferences.values();
- }
-
public void persist(FileStore lobStore) throws TeiidComponentException {
- ArrayList<Streamable<?>> lobs = new
ArrayList<Streamable<?>>(this.lobReferences.values());
- for (Streamable<?> lob:lobs) {
- persist(lob.getReferenceStreamId(), lobStore);
+ // stream the contents of lob into file store.
+ byte[] bytes = new byte[102400]; // 100k
+
+ for (Map.Entry<String, Streamable<?>> entry :
this.lobReferences.entrySet()) {
+ entry.setValue(persistLob(entry.getValue(), lobStore, bytes));
}
}
- public Streamable<?> persist(String id, FileStore fs) throws
TeiidComponentException {
- Streamable<?> persistedLob = this.lobFilestores.get(id);
- if (persistedLob == null) {
- Streamable<?> lobReference = this.lobReferences.get(id);
- if (lobReference == null) {
- throw new
TeiidComponentException(QueryPlugin.Util.getString("ProcessWorker.wrongdata"));
//$NON-NLS-1$
- }
-
- persistedLob = persistLob(lobReference, fs);
- synchronized (this) {
- this.lobFilestores.put(id, persistedLob);
- this.lobReferences.remove(id);
- }
- }
- return persistedLob;
- }
-
- private Streamable<?> persistLob(final Streamable<?> lob, final FileStore
store) throws TeiidComponentException {
- long offset = store.getLength();
- int length = 0;
- Streamable<?> persistedLob;
+ private Streamable<?> persistLob(final Streamable<?> lob, final FileStore
store, byte[] bytes) throws TeiidComponentException {
- // if this is XML and already saved to disk just return
- if (lob.getReference() instanceof SQLXMLImpl) {
+ // if this is already saved to disk just return
+ if (lob.getReference() instanceof BaseLob) {
try {
- SQLXMLImpl xml = (SQLXMLImpl)lob.getReference();
- InputStreamFactory isf = xml.getStreamFactory();
- if (isf instanceof FileStoreInputStreamFactory) {
+ BaseLob baseLob = (BaseLob)lob.getReference();
+ InputStreamFactory isf = baseLob.getStreamFactory();
+ if (isf.isPersistent()) {
return lob;
}
} catch (SQLException e) {
// go through regular persistence.
}
}
+ long offset = store.getLength();
+ int length = 0;
+ Streamable<?> persistedLob;
- // stream the contents of lob into file store.
- byte[] bytes = new byte[102400]; // 100k
try {
InputStreamFactory isf = new InputStreamFactory() {
@Override
@@ -178,20 +140,11 @@
}
};
InputStream is = isf.getInputStream();
- OutputStream fsos = new BufferedOutputStream(store.createOutputStream(),
IO_BUFFER_SIZE);
- while(true) {
- int read = is.read(bytes, 0, 102400);
- if (read == -1) {
- break;
- }
- length += read;
- fsos.write(bytes, 0, read);
- }
- fsos.close();
- is.close();
+ OutputStream fsos = store.createOutputStream();
+ 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;
@@ -201,6 +154,11 @@
public InputStream getInputStream() throws IOException {
return store.createInputStream(lobOffset, lobLength);
}
+
+ @Override
+ public boolean isPersistent() {
+ return true;
+ }
};
try {
Modified: trunk/engine/src/main/java/org/teiid/common/buffer/TupleBuffer.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/common/buffer/TupleBuffer.java 2010-12-24
02:09:29 UTC (rev 2802)
+++ trunk/engine/src/main/java/org/teiid/common/buffer/TupleBuffer.java 2010-12-24
18:47:11 UTC (rev 2803)
@@ -160,7 +160,6 @@
public void persistLobs() throws TeiidComponentException {
if (this.lobManager != null) {
- saveBatch(true, true);
this.lobManager.persist(this.lobStore);
}
}
@@ -236,6 +235,9 @@
if (LogManager.isMessageToBeRecorded(LogConstants.CTX_BUFFER_MGR,
MessageLevel.DETAIL)) {
LogManager.logDetail(LogConstants.CTX_BUFFER_MGR, "Removing
TupleBuffer:", this.tupleSourceID); //$NON-NLS-1$
}
+ if (this.lobStore != null) {
+ this.lobStore.remove();
+ }
this.batchBuffer = null;
purge();
this.manager.remove();
Modified: trunk/engine/src/main/java/org/teiid/common/buffer/impl/BufferManagerImpl.java
===================================================================
---
trunk/engine/src/main/java/org/teiid/common/buffer/impl/BufferManagerImpl.java 2010-12-24
02:09:29 UTC (rev 2802)
+++
trunk/engine/src/main/java/org/teiid/common/buffer/impl/BufferManagerImpl.java 2010-12-24
18:47:11 UTC (rev 2803)
@@ -33,7 +33,6 @@
import java.lang.ref.SoftReference;
import java.lang.ref.WeakReference;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.Iterator;
@@ -61,7 +60,6 @@
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.util.Assertion;
import org.teiid.logging.LogConstants;
import org.teiid.logging.LogManager;
@@ -318,10 +316,6 @@
if (lobManager != null) {
for (List<?> tuple : batch.getTuples()) {
lobManager.updateReferences(batchManager.lobIndexes, tuple);
- Collection<Streamable<?>> lobs = lobManager.getLobReferences();
- for(Streamable<?> lob: lobs) {
- lobManager.persist(lob.getReferenceStreamId(), batchManager.store);
- }
}
}
synchronized (batchManager.store) {
@@ -368,9 +362,6 @@
if (info != null) {
batchManager.unusedSpace.addAndGet(info[1]);
}
- if (lobManager != null) {
- lobManager.clear();
- }
activeBatch = null;
batchReference = null;
}
Modified: trunk/engine/src/main/java/org/teiid/dqp/internal/process/RequestWorkItem.java
===================================================================
---
trunk/engine/src/main/java/org/teiid/dqp/internal/process/RequestWorkItem.java 2010-12-24
02:09:29 UTC (rev 2802)
+++
trunk/engine/src/main/java/org/teiid/dqp/internal/process/RequestWorkItem.java 2010-12-24
18:47:11 UTC (rev 2803)
@@ -294,14 +294,12 @@
rowcount = resultsBuffer.getRowCount();
if (this.cid == null || !this.doneProducingBatches) {
resultsBuffer.remove();
- }
-
- try {
- if (cid != null && this.resultsBuffer.isLobs()) {
+ } else {
+ try {
this.resultsBuffer.persistLobs();
+ } catch (TeiidComponentException e) {
+ LogManager.logDetail(LogConstants.CTX_DQP,
QueryPlugin.Util.getString("failed_to_cache")); //$NON-NLS-1$
}
- } catch (TeiidComponentException e) {
- LogManager.logDetail(LogConstants.CTX_DQP,
QueryPlugin.Util.getString("failed_to_cache")); //$NON-NLS-1$
}
for (DataTierTupleSource connectorRequest : this.connectorInfo.values()) {
Modified:
trunk/engine/src/main/java/org/teiid/query/optimizer/relational/RelationalPlanner.java
===================================================================
---
trunk/engine/src/main/java/org/teiid/query/optimizer/relational/RelationalPlanner.java 2010-12-24
02:09:29 UTC (rev 2802)
+++
trunk/engine/src/main/java/org/teiid/query/optimizer/relational/RelationalPlanner.java 2010-12-24
18:47:11 UTC (rev 2803)
@@ -39,6 +39,7 @@
import org.teiid.api.exception.query.QueryValidatorException;
import org.teiid.client.plan.Annotation;
import org.teiid.client.plan.Annotation.Priority;
+import org.teiid.common.buffer.LobManager;
import org.teiid.core.TeiidComponentException;
import org.teiid.core.TeiidProcessingException;
import org.teiid.core.id.IDGenerator;
@@ -582,7 +583,7 @@
if (container instanceof StoredProcedure) {
boolean noCache = isNoCacheGroup(metadata, ((StoredProcedure)
container).getProcedureID(), option);
if (!noCache) {
- if (container.areResultsCachable() &&
Query.areResultsCachable(container.getProcedureParameters().keySet()) &&
context.isResultSetCacheEnabled()) {
+ if (context.isResultSetCacheEnabled() && container.areResultsCachable()
&& LobManager.getLobIndexes(new
ArrayList<ElementSymbol>(container.getProcedureParameters().keySet())) == null) {
container.getGroup().setGlobalTable(true);
container.setCacheHint(c.getCacheHint());
recordAnnotation(analysisRecord, Annotation.CACHED_PROCEDURE, Priority.LOW,
"SimpleQueryResolver.procedure_cache_used", container.getGroup());
//$NON-NLS-1$
Modified: trunk/engine/src/main/java/org/teiid/query/processor/xml/XMLUtil.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/query/processor/xml/XMLUtil.java 2010-12-24
02:09:29 UTC (rev 2802)
+++ trunk/engine/src/main/java/org/teiid/query/processor/xml/XMLUtil.java 2010-12-24
18:47:11 UTC (rev 2803)
@@ -86,6 +86,11 @@
public void free() throws IOException {
lobBuffer.remove();
}
+
+ @Override
+ public boolean isPersistent() {
+ return true;
+ }
}
/**