11:19 a.m.
Author: shawkins
Date: 2011-04-14 12:19:52 -0400 (Thu, 14 Apr 2011)
New Revision: 3090
Modified:
trunk/engine/src/main/java/org/teiid/common/buffer/FileStore.java
trunk/engine/src/main/java/org/teiid/common/buffer/FileStoreInputStreamFactory.java
trunk/engine/src/main/java/org/teiid/query/function/aggregate/TextAgg.java
trunk/engine/src/main/java/org/teiid/query/function/aggregate/XMLAgg.java
trunk/engine/src/main/java/org/teiid/query/function/source/XMLSystemFunctions.java
Log:
TEIID-1556 better handling for xmlagg and reducing the memory held by associated buffers
Modified: trunk/engine/src/main/java/org/teiid/common/buffer/FileStore.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/common/buffer/FileStore.java 2011-04-13 20:34:39
UTC (rev 3089)
+++ trunk/engine/src/main/java/org/teiid/common/buffer/FileStore.java 2011-04-14 16:19:52
UTC (rev 3090)
@@ -27,6 +27,7 @@
import java.io.OutputStream;
import java.lang.ref.PhantomReference;
import java.lang.ref.ReferenceQueue;
+import java.util.Arrays;
import java.util.Collections;
import java.util.IdentityHashMap;
import java.util.Set;
@@ -46,6 +47,7 @@
private byte[] buffer;
private int count;
private boolean bytesWritten;
+ private boolean closed;
public FileStoreOutputStream(int size) {
this.buffer = new byte[size];
@@ -58,6 +60,7 @@
@Override
public void write(byte[] b, int off, int len) throws IOException {
+ checkOpen();
if (len > buffer.length) {
flushBuffer();
writeDirect(b, off, len);
@@ -86,12 +89,18 @@
}
public void flushBuffer() throws IOException {
+ checkOpen();
if (count > 0) {
writeDirect(buffer, 0, count);
count = 0;
}
}
+ /**
+ * Return the buffer. Can be null if closed and the underlying filestore
+ * has been writen to.
+ * @return
+ */
public byte[] getBuffer() {
return buffer;
}
@@ -114,8 +123,21 @@
@Override
public void close() throws IOException {
flush();
+ closed = true;
+ if (bytesWritten) {
+ this.buffer = null;
+ } else {
+ //truncate
+ this.buffer = Arrays.copyOf(this.buffer, this.count);
+ }
}
+ private void checkOpen() {
+ if (closed) {
+ throw new IllegalStateException("Alread closed"); //$NON-NLS-1$
+ }
+ }
+
}
static class CleanupReference extends PhantomReference<Object> {
Modified:
trunk/engine/src/main/java/org/teiid/common/buffer/FileStoreInputStreamFactory.java
===================================================================
---
trunk/engine/src/main/java/org/teiid/common/buffer/FileStoreInputStreamFactory.java 2011-04-13
20:34:39 UTC (rev 3089)
+++
trunk/engine/src/main/java/org/teiid/common/buffer/FileStoreInputStreamFactory.java 2011-04-14
16:19:52 UTC (rev 3090)
@@ -59,10 +59,20 @@
return lobBuffer.getLength();
}
+ /**
+ * Returns a new writer instance that is backed by the shared output stream.
+ * Closing a writer will prevent further writes.
+ * @return
+ */
public Writer getWriter() {
return new OutputStreamWriter(getOuputStream(), Charset.forName(encoding));
}
+ /**
+ * The returned output stream is shared among all uses.
+ * Once closed no further writing can occur
+ * @return
+ */
public FileStoreOutputStream getOuputStream() {
if (fsos == null) {
fsos = lobBuffer.createOutputStream(DataTypeManager.MAX_LOB_MEMORY_BYTES);
@@ -77,6 +87,6 @@
@Override
public boolean isPersistent() {
- return true;
+ return fsos == null || fsos.bytesWritten();
}
}
\ No newline at end of file
Modified: trunk/engine/src/main/java/org/teiid/query/function/aggregate/TextAgg.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/query/function/aggregate/TextAgg.java 2011-04-13
20:34:39 UTC (rev 3089)
+++ trunk/engine/src/main/java/org/teiid/query/function/aggregate/TextAgg.java 2011-04-14
16:19:52 UTC (rev 3090)
@@ -72,7 +72,7 @@
}
}, textLine.getDelimiter(), textLine.getQuote()));
}
- w.close();
+ w.flush();
return fisf;
} catch (IOException e) {
throw new TeiidProcessingException(e);
@@ -96,7 +96,7 @@
String in = (String)input;
Writer w = result.getWriter();
w.write(in);
- w.close();
+ w.flush();
} catch (IOException e) {
throw new TeiidProcessingException(e);
}
Modified: trunk/engine/src/main/java/org/teiid/query/function/aggregate/XMLAgg.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/query/function/aggregate/XMLAgg.java 2011-04-13
20:34:39 UTC (rev 3089)
+++ trunk/engine/src/main/java/org/teiid/query/function/aggregate/XMLAgg.java 2011-04-14
16:19:52 UTC (rev 3090)
@@ -27,7 +27,7 @@
import org.teiid.core.TeiidComponentException;
import org.teiid.core.TeiidProcessingException;
import org.teiid.core.types.XMLType;
-import org.teiid.query.function.source.XMLSystemFunctions;
+import org.teiid.query.function.source.XMLSystemFunctions.XmlConcat;
import org.teiid.query.util.CommandContext;
/**
@@ -35,7 +35,8 @@
*/
public class XMLAgg extends AggregateFunction {
- private XMLType result;
+ private XMLType result;
+ private XmlConcat concat;
private CommandContext context;
public XMLAgg(CommandContext context) {
@@ -43,7 +44,8 @@
}
public void reset() {
- result = null;
+ concat = null;
+ result = null;
}
/**
@@ -52,13 +54,25 @@
* @see org.teiid.query.function.aggregate.AggregateFunction#addInputDirect(Object,
List)
*/
public void addInputDirect(Object input, List<?> tuple) throws
TeiidComponentException, TeiidProcessingException {
- result = XMLSystemFunctions.xmlConcat(context, result, input);
+ if (concat == null) {
+ concat = new XmlConcat(context.getBufferManager());
+ }
+ concat.addValue(input);
}
/**
+ * @throws TeiidProcessingException
+ * @throws TeiidComponentException
* @see org.teiid.query.function.aggregate.AggregateFunction#getResult()
*/
- public Object getResult() {
+ public Object getResult() throws TeiidComponentException, TeiidProcessingException {
+ if (result == null) {
+ if (concat == null) {
+ return null;
+ }
+ result = concat.close();
+ concat = null;
+ }
return result;
}
Modified:
trunk/engine/src/main/java/org/teiid/query/function/source/XMLSystemFunctions.java
===================================================================
---
trunk/engine/src/main/java/org/teiid/query/function/source/XMLSystemFunctions.java 2011-04-13
20:34:39 UTC (rev 3089)
+++
trunk/engine/src/main/java/org/teiid/query/function/source/XMLSystemFunctions.java 2011-04-14
16:19:52 UTC (rev 3090)
@@ -386,27 +386,67 @@
return singleValue;
}
- XMLType result = new
XMLType(XMLSystemFunctions.saveToBufferManager(context.getBufferManager(), new
XMLTranslator() {
-
- @Override
- public void translate(Writer writer) throws TransformerException,
- IOException {
- try {
- XMLOutputFactory factory = getOutputFactory();
- XMLEventWriter eventWriter = factory.createXMLEventWriter(writer);
- XMLEventFactory eventFactory = XMLEventFactory.newInstance();
- convertValue(writer, eventWriter, eventFactory, xml);
- for (Object object : other) {
- convertValue(writer, eventWriter, eventFactory, object);
- }
- eventWriter.flush(); //woodstox needs a flush rather than a close
- } catch (XMLStreamException e) {
- throw new TransformerException(e);
- }
+ XmlConcat concat = new XmlConcat(context.getBufferManager());
+ concat.addValue(xml);
+ for (Object object : other) {
+ concat.addValue(object);
+ }
+ return concat.close();
+ }
+
+ public static class XmlConcat {
+ private XMLOutputFactory factory;
+ private XMLEventWriter eventWriter;
+ private XMLEventFactory eventFactory;
+ private Writer writer;
+ private FileStoreInputStreamFactory fsisf;
+ private FileStore fs;
+
+ public XmlConcat(BufferManager bm) throws TeiidProcessingException {
+ fs = bm.createFileStore("xml"); //$NON-NLS-1$
+ fsisf = new FileStoreInputStreamFactory(fs, Streamable.ENCODING);
+ writer = fsisf.getWriter();
+ factory = getOutputFactory();
+ try {
+ eventWriter = factory.createXMLEventWriter(writer);
+ } catch (XMLStreamException e) {
+ fs.remove();
+ throw new TeiidProcessingException(e);
}
- }));
- result.setType(Type.CONTENT);
- return result;
+ eventFactory = XMLEventFactory.newInstance();
+ }
+
+ public void addValue(Object object) throws TeiidProcessingException {
+ try {
+ convertValue(writer, eventWriter, eventFactory, object);
+ } catch (IOException e) {
+ fs.remove();
+ throw new TeiidProcessingException(e);
+ } catch (XMLStreamException e) {
+ fs.remove();
+ throw new TeiidProcessingException(e);
+ } catch (TransformerException e) {
+ fs.remove();
+ throw new TeiidProcessingException(e);
+ }
+ }
+
+ public XMLType close() throws TeiidProcessingException {
+ try {
+ eventWriter.flush();
+ writer.close();
+ } catch (XMLStreamException e) {
+ fs.remove();
+ throw new TeiidProcessingException(e);
+ } catch (IOException e) {
+ fs.remove();
+ throw new TeiidProcessingException(e);
+ }
+ XMLType result = new XMLType(new SQLXMLImpl(fsisf));
+ result.setType(Type.CONTENT);
+ return result;
+ }
+
}
private static XMLOutputFactory getOutputFactory() throws FactoryConfigurationError {