[teiid-commits] teiid SVN: r4571 - in branches/7.7.x: client/src/main/java/org/teiid/client and 8 other directories.

Author: jolee Date: 2013-05-23 15:23:42 -0400 (Thu, 23 May 2013) New Revision: 4571 Modified: branches/7.7.x/build/kits/jboss-container/teiid-releasenotes.html branches/7.7.x/client/src/main/java/org/teiid/client/RequestMessage.java branches/7.7.x/client/src/main/java/org/teiid/client/ResultsMessage.java branches/7.7.x/client/src/main/java/org/teiid/jdbc/ResultSetImpl.java branches/7.7.x/client/src/main/java/org/teiid/jdbc/StatementImpl.java branches/7.7.x/client/src/main/java/org/teiid/net/ServerConnection.java branches/7.7.x/client/src/main/java/org/teiid/net/socket/SocketServerConnection.java branches/7.7.x/client/src/test/java/org/teiid/client/TestRequestMessage.java branches/7.7.x/client/src/test/java/org/teiid/jdbc/TestResultSet.java branches/7.7.x/engine/src/main/java/org/teiid/dqp/internal/process/RequestWorkItem.java branches/7.7.x/engine/src/test/java/org/teiid/dqp/message/TestResultsMessage.java branches/7.7.x/runtime/src/main/java/org/teiid/transport/LocalServerConnection.java Log: TEIID-2504: Improve socket results processing Modified: branches/7.7.x/build/kits/jboss-container/teiid-releasenotes.html =================================================================== --- branches/7.7.x/build/kits/jboss-container/teiid-releasenotes.html 2013-05-22 18:29:09 UTC (rev 4570) +++ branches/7.7.x/build/kits/jboss-container/teiid-releasenotes.html 2013-05-23 19:23:42 UTC (rev 4571) @@ -256,6 +256,8 @@ </li> <li>[<a href='https://issues.jboss.org/browse/TEIID-2491'>TEIID-2491&l...>] - Issue with date format parsing using ODBC Driver (EDS Specific BZ961071) </li> +<li>[<a href='https://issues.jboss.org/browse/TEIID-2504'>TEIID-2504&l...>] - Improve socket results processing +</li> </ul> <h4>From 7.7.6</h4> Modified: branches/7.7.x/client/src/main/java/org/teiid/client/RequestMessage.java =================================================================== --- branches/7.7.x/client/src/main/java/org/teiid/client/RequestMessage.java 2013-05-22 18:29:09 UTC (rev 4570) +++ branches/7.7.x/client/src/main/java/org/teiid/client/RequestMessage.java 2013-05-23 19:23:42 UTC (rev 4571) @@ -22,11 +22,14 @@ package org.teiid.client; +import java.io.EOFException; import java.io.Externalizable; import java.io.IOException; import java.io.ObjectInput; import java.io.ObjectOutput; +import java.io.OptionalDataException; import java.io.Serializable; +import java.sql.ResultSet; import java.util.Arrays; import java.util.Collections; import java.util.List; @@ -90,6 +93,8 @@ private int transactionIsolation; private boolean noExec; private transient boolean sync; + + private boolean delaySerialization = true; public RequestMessage() { } @@ -177,6 +182,7 @@ */ public void setCursorType(int cursorType) { this.cursorType = cursorType; + this.delaySerialization = this.cursorType == ResultSet.TYPE_FORWARD_ONLY; } /** @@ -378,6 +384,13 @@ this.executionId = in.readLong(); this.transactionIsolation = in.readInt(); this.noExec = in.readBoolean(); + try { + byte options = in.readByte(); + //8.4 property + this.delaySerialization = (options & 2) == 2; + } catch (OptionalDataException e) { + } catch (EOFException e) { + } } @Override @@ -402,6 +415,20 @@ out.writeLong(executionId); out.writeInt(transactionIsolation); out.writeBoolean(noExec); + byte options = 0; + if (delaySerialization) { + options |= 2; + } + out.writeByte(options); } + public boolean isDelaySerialization() { + return delaySerialization; + } + + //TODO: allow for the client to explicitly enable/disable this behavior + public void setDelaySerialization(boolean delaySerialization) { + this.delaySerialization = delaySerialization; + } + } Modified: branches/7.7.x/client/src/main/java/org/teiid/client/ResultsMessage.java =================================================================== --- branches/7.7.x/client/src/main/java/org/teiid/client/ResultsMessage.java 2013-05-22 18:29:09 UTC (rev 4570) +++ branches/7.7.x/client/src/main/java/org/teiid/client/ResultsMessage.java 2013-05-23 19:23:42 UTC (rev 4571) @@ -22,6 +22,7 @@ package org.teiid.client; +import java.io.ByteArrayInputStream; import java.io.Externalizable; import java.io.IOException; import java.io.ObjectInput; @@ -35,7 +36,11 @@ import org.teiid.client.plan.PlanNode; import org.teiid.client.util.ExceptionHolder; import org.teiid.core.TeiidException; +import org.teiid.core.util.AccessibleByteArrayOutputStream; import org.teiid.core.util.ExternalizeUtil; +import org.teiid.jdbc.TeiidSQLException; +import org.teiid.netty.handler.codec.serialization.CompactObjectInputStream; +import org.teiid.netty.handler.codec.serialization.CompactObjectOutputStream; /** @@ -83,6 +88,9 @@ private Collection<Annotation> annotations; private boolean isUpdateResult; + + private boolean delayDeserialization; + byte[] resultBytes; public ResultsMessage(){ } @@ -107,6 +115,21 @@ public List<?>[] getResults() { return results.toArray(new List[results.size()]); } + + public void processResults() throws TeiidSQLException { + if (results == null && resultBytes != null) { + try { + CompactObjectInputStream ois = new CompactObjectInputStream(new ByteArrayInputStream(resultBytes), ResultsMessage.class.getClassLoader()); + results = BatchSerializer.readBatch(ois, dataTypes, (byte)0); + } catch (IOException e) { + throw TeiidSQLException.create(e); + } catch (ClassNotFoundException e) { + throw TeiidSQLException.create(e); + } finally { + resultBytes = null; + } + } + } public void setResults(List<?>[] results) { this.results = Arrays.asList(results); @@ -251,6 +274,14 @@ if (holder != null) { this.exception = (TeiidException)holder.getException(); } + + //delayed deserialization + if (results == null && this.exception == null) { + int length = in.readInt(); + resultBytes = new byte[length]; + in.readFully(resultBytes); + } + List<ExceptionHolder> holderList = (List<ExceptionHolder>)in.readObject(); if (holderList != null) { this.warnings = ExceptionHolder.toThrowables(holderList); @@ -274,7 +305,11 @@ ExternalizeUtil.writeArray(out, dataTypes); // Results data - BatchSerializer.writeBatch(out, dataTypes, results, (byte)0); + if (delayDeserialization) { + BatchSerializer.writeBatch(out, dataTypes, null, (byte)0); + } else { + BatchSerializer.writeBatch(out, dataTypes, results, (byte)0); + } // Plan descriptions out.writeObject(this.planDescription); @@ -284,6 +319,16 @@ } else { out.writeObject(exception); } + + if (delayDeserialization && results != null) { + AccessibleByteArrayOutputStream baos = new AccessibleByteArrayOutputStream(1 << 13); + CompactObjectOutputStream oos = new CompactObjectOutputStream(baos); + BatchSerializer.writeBatch(oos, dataTypes, results, (byte)0); + oos.close(); + out.writeInt(baos.getCount()); + out.write(baos.getBuffer(), 0, baos.getCount()); + } + if (this.warnings != null) { out.writeObject(ExceptionHolder.toExceptionHolders(this.warnings)); } else { @@ -357,5 +402,9 @@ public void setClientSerializationVersion(byte clientSerializationVersion) { this.clientSerializationVersion = clientSerializationVersion; } + + public void setDelayDeserialization(boolean delayDeserialization) { + this.delayDeserialization = delayDeserialization; + } } Modified: branches/7.7.x/client/src/main/java/org/teiid/jdbc/ResultSetImpl.java =================================================================== --- branches/7.7.x/client/src/main/java/org/teiid/jdbc/ResultSetImpl.java 2013-05-22 18:29:09 UTC (rev 4570) +++ branches/7.7.x/client/src/main/java/org/teiid/jdbc/ResultSetImpl.java 2013-05-23 19:23:42 UTC (rev 4571) @@ -98,6 +98,9 @@ private int fetchSize; private Map<String, Integer> columnMap; + + private ResultsFuture<ResultsMessage> prefetch; + private boolean usePrefetch; /** * Constructor. @@ -117,7 +120,6 @@ // server latency-related timestamp this.requestID = statement.getCurrentRequestID(); this.cursorType = statement.getResultSetType(); - this.batchResults = new BatchResults(this, getCurrentBatch(resultsMsg), this.cursorType == ResultSet.TYPE_FORWARD_ONLY ? 1 : BatchResults.DEFAULT_SAVED_BATCHES); this.serverTimeZone = statement.getServerTimeZone(); if (metadata == null) { MetadataProvider provider = new DeferredMetadataProvider(resultsMsg.getColumnNames(), @@ -138,6 +140,8 @@ if (logger.isLoggable(Level.FINER)) { logger.finer("Creating ResultSet requestID: " + requestID + " beginRow: " + resultsMsg.getFirstRow() + " resultsColumns: " + resultColumns + " parameters: " + parameters); //$NON-NLS-1$ //$NON-NLS-2$ //$NON-NLS-3$ //$NON-NLS-4$ } + this.usePrefetch = cursorType == ResultSet.TYPE_FORWARD_ONLY && !statement.useCallingThread(); + this.batchResults = new BatchResults(this, getCurrentBatch(resultsMsg), this.cursorType == ResultSet.TYPE_FORWARD_ONLY ? 1 : BatchResults.DEFAULT_SAVED_BATCHES); } public void setMaxFieldSize(int maxFieldSize) { @@ -353,6 +357,17 @@ public Batch requestBatch(int beginRow) throws SQLException{ checkClosed(); try { + if (prefetch != null) { + //TODO: this is not efficient if the user is skipping around the results + //but the server logic at this point basically requires us + //to read what we have requested before requesting more (no queuing) + ResultsMessage result = getResults(prefetch); + prefetch = null; + Batch nextBatch = processBatch(result); + if (result.getFirstRow() == beginRow) { + return nextBatch; + } + } ResultsFuture<ResultsMessage> results = submitRequestBatch(beginRow); ResultsMessage currentResultMsg = getResults(results); return processBatch(currentResultMsg); @@ -400,8 +415,14 @@ return currentResultMsg; } - private Batch getCurrentBatch(ResultsMessage currentResultMsg) { + private Batch getCurrentBatch(ResultsMessage currentResultMsg) throws TeiidSQLException { this.updatedPlanDescription = currentResultMsg.getPlanDescription(); + if (usePrefetch + && prefetch == null && currentResultMsg.getLastRow() != currentResultMsg.getFinalRow()) { + //fetch before processing the results + prefetch = submitRequestBatch(currentResultMsg.getLastRow() + 1); + } + currentResultMsg.processResults(); boolean isLast = currentResultMsg.getResultsList().size() == 0 || currentResultMsg.getFinalRow() == currentResultMsg.getLastRow(); Batch result = new Batch(currentResultMsg.getResults(), currentResultMsg.getFirstRow(), currentResultMsg.getLastRow(), isLast); result.setLastRow(currentResultMsg.getFinalRow()); @@ -1689,6 +1710,10 @@ public void updateRowId(String columnLabel, RowId x) throws SQLException { throw SqlUtil.createFeatureNotSupportedException(); } + + ResultsFuture<ResultsMessage> getPrefetch() { + return prefetch; + } public void updateShort(int columnIndex, short x) throws SQLException { throw SqlUtil.createFeatureNotSupportedException(); Modified: branches/7.7.x/client/src/main/java/org/teiid/jdbc/StatementImpl.java =================================================================== --- branches/7.7.x/client/src/main/java/org/teiid/jdbc/StatementImpl.java 2013-05-22 18:29:09 UTC (rev 4570) +++ branches/7.7.x/client/src/main/java/org/teiid/jdbc/StatementImpl.java 2013-05-23 19:23:42 UTC (rev 4571) @@ -588,8 +588,7 @@ reqMsg.setFetchSize(this.fetchSize); reqMsg.setRowLimit(this.maxRows); reqMsg.setTransactionIsolation(this.driverConnection.getTransactionIsolation()); - String useCallingThread = getExecutionProperty(EmbeddedProfile.USE_CALLING_THREAD); - reqMsg.setSync(synch && (useCallingThread == null || Boolean.valueOf(useCallingThread))); + reqMsg.setSync(synch && useCallingThread()); // Get connection properties and set them onto request message copyPropertiesToRequest(reqMsg); @@ -617,6 +616,14 @@ } return pendingResult; } + + boolean useCallingThread() throws SQLException { + if (this.getConnection().getServerConnection() == null || !this.getConnection().getServerConnection().isLocal()) { + return false; + } + String useCallingThread = getExecutionProperty(EmbeddedProfile.USE_CALLING_THREAD); + return (useCallingThread == null || Boolean.valueOf(useCallingThread)); + } public static ResultsFuture<Boolean> booleanFuture(boolean isTrue) { ResultsFuture<Boolean> rs = new ResultsFuture<Boolean>(); Modified: branches/7.7.x/client/src/main/java/org/teiid/net/ServerConnection.java =================================================================== --- branches/7.7.x/client/src/main/java/org/teiid/net/ServerConnection.java 2013-05-22 18:29:09 UTC (rev 4570) +++ branches/7.7.x/client/src/main/java/org/teiid/net/ServerConnection.java 2013-05-23 19:23:42 UTC (rev 4571) @@ -42,4 +42,6 @@ void authenticate() throws ConnectionException, CommunicationException; + boolean isLocal(); + } \ No newline at end of file Modified: branches/7.7.x/client/src/main/java/org/teiid/net/socket/SocketServerConnection.java =================================================================== --- branches/7.7.x/client/src/main/java/org/teiid/net/socket/SocketServerConnection.java 2013-05-22 18:29:09 UTC (rev 4570) +++ branches/7.7.x/client/src/main/java/org/teiid/net/socket/SocketServerConnection.java 2013-05-23 19:23:42 UTC (rev 4571) @@ -372,4 +372,9 @@ } } } + + @Override + public boolean isLocal() { + return false; + } } \ No newline at end of file Modified: branches/7.7.x/client/src/test/java/org/teiid/client/TestRequestMessage.java =================================================================== --- branches/7.7.x/client/src/test/java/org/teiid/client/TestRequestMessage.java 2013-05-22 18:29:09 UTC (rev 4570) +++ branches/7.7.x/client/src/test/java/org/teiid/client/TestRequestMessage.java 2013-05-23 19:23:42 UTC (rev 4571) @@ -25,15 +25,14 @@ import java.util.ArrayList; import java.util.List; -import org.teiid.client.RequestMessage; +import junit.framework.TestCase; + import org.teiid.client.RequestMessage.ShowPlan; import org.teiid.client.RequestMessage.StatementType; import org.teiid.core.TeiidProcessingException; import org.teiid.core.util.UnitTestUtil; -import junit.framework.TestCase; - public class TestRequestMessage extends TestCase { /** @@ -91,7 +90,7 @@ assertEquals("xMLFormat", copy.getXMLFormat()); //$NON-NLS-1$ assertEquals(ShowPlan.ON, copy.getShowPlan()); assertEquals(1313, copy.getRowLimit()); - + assertTrue(copy.isDelaySerialization()); } public void testInvalidTxnAutoWrap() { Modified: branches/7.7.x/client/src/test/java/org/teiid/jdbc/TestResultSet.java =================================================================== --- branches/7.7.x/client/src/test/java/org/teiid/jdbc/TestResultSet.java 2013-05-22 18:29:09 UTC (rev 4570) +++ branches/7.7.x/client/src/test/java/org/teiid/jdbc/TestResultSet.java 2013-05-23 19:23:42 UTC (rev 4571) @@ -676,9 +676,15 @@ } @Test public void testForwardOnly() throws Exception { - ResultSetImpl cs = helpExecuteQuery(400, 1000, ResultSet.TYPE_FORWARD_ONLY); - + ResultSetImpl cs = helpExecuteQuery(400, 1300, ResultSet.TYPE_FORWARD_ONLY); + int i = 0; while (cs.next()) { + i++; + if (i <= 1200) { + assertNotNull(cs.getPrefetch()); + } else { + assertNull(cs.getPrefetch()); + } cs.getObject(1); } Modified: branches/7.7.x/engine/src/main/java/org/teiid/dqp/internal/process/RequestWorkItem.java =================================================================== --- branches/7.7.x/engine/src/main/java/org/teiid/dqp/internal/process/RequestWorkItem.java 2013-05-22 18:29:09 UTC (rev 4570) +++ branches/7.7.x/engine/src/main/java/org/teiid/dqp/internal/process/RequestWorkItem.java 2013-05-23 19:23:42 UTC (rev 4571) @@ -34,17 +34,18 @@ import java.util.concurrent.Callable; import org.teiid.client.RequestMessage; +import org.teiid.client.RequestMessage.ShowPlan; +import org.teiid.client.ResizingArrayList; import org.teiid.client.ResultsMessage; import org.teiid.client.SourceWarning; -import org.teiid.client.RequestMessage.ShowPlan; import org.teiid.client.lob.LobChunk; import org.teiid.client.metadata.ParameterInfo; import org.teiid.client.util.ResultsReceiver; import org.teiid.client.xa.XATransactionException; import org.teiid.common.buffer.BlockedException; +import org.teiid.common.buffer.BufferManager.TupleSourceType; import org.teiid.common.buffer.TupleBatch; import org.teiid.common.buffer.TupleBuffer; -import org.teiid.common.buffer.BufferManager.TupleSourceType; import org.teiid.core.TeiidComponentException; import org.teiid.core.TeiidException; import org.teiid.core.TeiidProcessingException; @@ -59,14 +60,14 @@ import org.teiid.dqp.message.AtomicRequestID; import org.teiid.dqp.message.RequestID; import org.teiid.dqp.service.TransactionContext; +import org.teiid.dqp.service.TransactionContext.Scope; import org.teiid.dqp.service.TransactionService; -import org.teiid.dqp.service.TransactionContext.Scope; +import org.teiid.jdbc.EnhancedTimer.Task; import org.teiid.jdbc.SQLStates; -import org.teiid.jdbc.EnhancedTimer.Task; +import org.teiid.logging.CommandLogMessage.Event; import org.teiid.logging.LogConstants; import org.teiid.logging.LogManager; import org.teiid.logging.MessageLevel; -import org.teiid.logging.CommandLogMessage.Event; import org.teiid.metadata.FunctionMethod.Determinism; import org.teiid.query.QueryPlugin; import org.teiid.query.analysis.AnalysisRecord; @@ -82,7 +83,8 @@ public class RequestWorkItem extends AbstractWorkItem implements PrioritizedRunnable { //TODO: this could be configurable - private static final int OUTPUT_BUFFER_MAX_BATCHES = 20; + private static final int OUTPUT_BUFFER_MAX_BATCHES = 8; + private static final int CLIENT_FETCH_MAX_BATCHES = 3; private final class WorkWrapper<T> implements DQPCore.CompletionListener<T> { @@ -628,16 +630,33 @@ //TODO: support fetching more than 1 batch boolean fromBuffer = false; + int count = this.end - this.begin + 1; if (batch == null || !(batch.containsRow(this.begin) || (batch.getTerminationFlag() && batch.getEndRow() <= this.begin))) { if (savedBatch != null && savedBatch.containsRow(this.begin)) { batch = savedBatch; } else { batch = resultsBuffer.getBatch(begin); + //fetch more than 1 batch from the buffer + boolean first = true; + int batches = CLIENT_FETCH_MAX_BATCHES; + for (int i = 1; i < batches && batch.getRowCount() + resultsBuffer.getBatchSize() <= count && !batch.getTerminationFlag(); i++) { + TupleBatch next = resultsBuffer.getBatch(batch.getEndRow() + 1); + if (next.getRowCount() == 0) { + break; + } + if (first) { + first = false; + TupleBatch old = batch; + batch = new TupleBatch(batch.getBeginRow(), new ResizingArrayList<List<?>>(batch.getTuples())); + batch.setTerminationFlag(old.getTerminationFlag()); + } + batch.getTuples().addAll(next.getTuples()); + batch.setTerminationFlag(next.getTerminationFlag()); + } } savedBatch = null; fromBuffer = true; } - int count = this.end - this.begin + 1; if (batch.getRowCount() > count) { int beginRow = Math.min(this.begin, batch.getEndRow() - count + 1); int endRow = Math.min(beginRow + count - 1, batch.getEndRow()); @@ -661,6 +680,9 @@ response.setUpdateResult(this.returnsUpdateCount); // set final row response.setFinalRow(finalRowCount); + if (response.getLastRow() == finalRowCount) { + response.setDelayDeserialization(false); + } // send any warnings with the response object List<Throwable> responseWarnings = new ArrayList<Throwable>(); @@ -707,6 +729,7 @@ ResultsMessage result = new ResultsMessage(batch, columnNames, dataTypes); result.setClientSerializationVersion((byte)(this.dqpWorkContext.getClientVersion().compareTo(Version.SEVEN_6) >= 0?1:0)); setAnalysisRecords(result); + result.setDelayDeserialization(this.requestMsg.isDelaySerialization() && this.originalCommand.returnsResultSet()); return result; } Modified: branches/7.7.x/engine/src/test/java/org/teiid/dqp/message/TestResultsMessage.java =================================================================== --- branches/7.7.x/engine/src/test/java/org/teiid/dqp/message/TestResultsMessage.java 2013-05-22 18:29:09 UTC (rev 4570) +++ branches/7.7.x/engine/src/test/java/org/teiid/dqp/message/TestResultsMessage.java 2013-05-23 19:23:42 UTC (rev 4571) @@ -114,11 +114,11 @@ assertNotNull(copy.getPlanDescription()); assertEquals(4, copy.getPlanDescription().getProperties().size()); - - assertNotNull(copy.getResults()); - assertEquals(1, copy.getResults().length); - assertNotNull(copy.getResults()[0]); - assertEquals(4, copy.getResults()[0].size()); + List<?>[] results = copy.getResults(); + assertNotNull(results); + assertEquals(1, results.length); + assertNotNull(results[0]); + assertEquals(4, results[0].size()); assertEquals(new BigInteger("100"), copy.getResults()[0].get(0)); //$NON-NLS-1$ assertEquals(new BigInteger("200"), copy.getResults()[0].get(1)); //$NON-NLS-1$ assertEquals(new BigInteger("300"), copy.getResults()[0].get(2)); //$NON-NLS-1$ @@ -130,7 +130,16 @@ assertEquals("warning1", ((Exception)copy.getWarnings().get(0)).getMessage()); //$NON-NLS-1$ assertEquals(Exception.class, copy.getWarnings().get(1).getClass()); assertEquals("warning2", ((Exception)copy.getWarnings().get(1)).getMessage()); //$NON-NLS-1$ + } + + public void testDelayedDeserialization() throws Exception { + ResultsMessage message = example(); + message.setDelayDeserialization(true); + ResultsMessage copy = UnitTestUtil.helpSerialize(message); + assertNull(copy.getResultsList()); + copy.processResults(); + assertNotNull(copy.getResultsList()); } } Modified: branches/7.7.x/runtime/src/main/java/org/teiid/transport/LocalServerConnection.java =================================================================== --- branches/7.7.x/runtime/src/main/java/org/teiid/transport/LocalServerConnection.java 2013-05-22 18:29:09 UTC (rev 4570) +++ branches/7.7.x/runtime/src/main/java/org/teiid/transport/LocalServerConnection.java 2013-05-23 19:23:42 UTC (rev 4571) @@ -188,4 +188,9 @@ public void cleanUp() { } + + @Override + public boolean isLocal() { + return true; + } }