[teiid-commits] teiid SVN: r4235 - in trunk: test-integration/common/src/test/java/org/teiid/jdbc and 1 other directory.

[teiid-commits at lists.jboss.org]

Author: shawkins
Date: 2012-07-13 09:03:44 -0400 (Fri, 13 Jul 2012)
New Revision: 4235

Modified:
   trunk/engine/src/main/java/org/teiid/dqp/internal/process/CachedResults.java
   trunk/engine/src/main/java/org/teiid/dqp/internal/process/RequestWorkItem.java
   trunk/test-integration/common/src/test/java/org/teiid/jdbc/TestResultsCache.java
Log:
TEIID-2098 ensuring maxrows does not cause invalid results with cache entries

Modified: trunk/engine/src/main/java/org/teiid/dqp/internal/process/CachedResults.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/dqp/internal/process/CachedResults.java	2012-07-12 18:24:56 UTC (rev 4234)
+++ trunk/engine/src/main/java/org/teiid/dqp/internal/process/CachedResults.java	2012-07-13 13:03:44 UTC (rev 4235)
@@ -53,6 +53,7 @@
 	private CacheHint hint;
 	private String uuid;
 	private boolean hasLobs;
+	private int rowLimit;
 	
 	private AccessInfo accessInfo = new AccessInfo();
 	
@@ -125,4 +126,12 @@
 		return accessInfo;
 	}
 	
+	public int getRowLimit() {
+		return rowLimit;
+	}
+	
+	public void setRowLimit(int rowLimit) {
+		this.rowLimit = rowLimit;
+	}
+	
 }

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	2012-07-12 18:24:56 UTC (rev 4234)
+++ trunk/engine/src/main/java/org/teiid/dqp/internal/process/RequestWorkItem.java	2012-07-13 13:03:44 UTC (rev 4235)
@@ -379,14 +379,7 @@
 			sendResultsIfNeeded(null);
 			this.resultsBuffer = collector.collectTuples();
 			if (!doneProducingBatches) {
-				doneProducingBatches();
-				//TODO: we could perform more tracking to know what source lobs are in use
-				if (this.resultsBuffer.getLobCount() == 0) {
-					for (DataTierTupleSource connectorRequest : getConnectorRequests()) {
-						connectorRequest.fullyCloseSource();
-				    }
-				}
-				addToCache();
+				done();
 			}
 		}
 		if (this.transactionState == TransactionState.ACTIVE) {
@@ -516,12 +509,14 @@
 		    	cachable = cacheId.setParameters(requestMsg.getParameterValues());
 				if (cachable) {
 					CachedResults cr = rsCache.get(cacheId);
-					if (cr != null) {
+					//check that there are enough cached results
+					//TODO: possibly ignore max rows for caching
+					if (cr != null && (cr.getRowLimit() == 0 || (requestMsg.getRowLimit() != 0 && requestMsg.getRowLimit() <= cr.getRowLimit()))) {
 						this.resultsBuffer = cr.getResults();
 						request.initMetadata();
 						this.originalCommand = cr.getCommand(requestMsg.getCommandString(), request.metadata, pi);
 						if (!request.validateAccess(requestMsg.getCommands(), this.originalCommand, CommandType.CACHED)) {
-							this.doneProducingBatches();
+							doneProducingBatches();
 							return;
 						}
 						LogManager.logDetail(LogConstants.CTX_DQP, requestID, "Cached result command to be modified, will not use the cached results", cacheId); //$NON-NLS-1$
@@ -551,8 +546,7 @@
 					super.flushBatchDirect(batch, add);
 				}
 				if (batch.getTerminationFlag()) {
-					doneProducingBatches();
-					addToCache();
+					done();
 				}
 				synchronized (lobStreams) {
 					if (resultsBuffer.isLobs()) {
@@ -618,6 +612,9 @@
     	CachedResults cr = new CachedResults();
     	cr.setCommand(originalCommand);
         cr.setResults(resultsBuffer, processor.getProcessorPlan());
+        if (requestMsg.getRowLimit() > 0 && resultsBuffer.getRowCount() == requestMsg.getRowLimit()) {
+        	cr.setRowLimit(resultsBuffer.getRowCount());
+        }
         if (originalCommand.getCacheHint() != null) {
         	LogManager.logDetail(LogConstants.CTX_DQP, requestID, "Using cache hint", originalCommand.getCacheHint()); //$NON-NLS-1$
 			resultsBuffer.setPrefersMemory(originalCommand.getCacheHint().isPrefersMemory());
@@ -973,6 +970,17 @@
 		}
 	}
 
+	private void done() {
+		doneProducingBatches();
+		//TODO: we could perform more tracking to know what source lobs are in use
+		if (this.resultsBuffer.getLobCount() == 0) {
+			for (DataTierTupleSource connectorRequest : getConnectorRequests()) {
+				connectorRequest.fullyCloseSource();
+		    }
+		}
+		addToCache();
+	}
+
 	private void doneProducingBatches() {
 		this.doneProducingBatches = true;
 		dqpCore.finishProcessing(this);

Modified: trunk/test-integration/common/src/test/java/org/teiid/jdbc/TestResultsCache.java
===================================================================
--- trunk/test-integration/common/src/test/java/org/teiid/jdbc/TestResultsCache.java	2012-07-12 18:24:56 UTC (rev 4234)
+++ trunk/test-integration/common/src/test/java/org/teiid/jdbc/TestResultsCache.java	2012-07-13 13:03:44 UTC (rev 4235)
@@ -26,9 +26,13 @@
 
 import java.sql.Connection;
 import java.sql.ResultSet;
+import java.sql.SQLException;
 import java.sql.Statement;
 
+import org.junit.After;
+import org.junit.AfterClass;
 import org.junit.Before;
+import org.junit.BeforeClass;
 import org.junit.Test;
 import org.teiid.core.util.UnitTestUtil;
 
@@ -37,13 +41,25 @@
 public class TestResultsCache {
 
 	private Connection conn;
+	private static FakeServer server;
+
+	@BeforeClass public static void oneTimeSetup() throws Exception {
+		server = new FakeServer(true);
+    	server.deployVDB("test", UnitTestUtil.getTestDataPath() + "/TestCase3473/test.vdb");
+	}
 	
+	@AfterClass public static void oneTimeTeardown() {
+		server.stop();
+	}
+	
 	@Before public void setUp() throws Exception {
-    	FakeServer server = new FakeServer(true);
-    	server.deployVDB("test", UnitTestUtil.getTestDataPath() + "/TestCase3473/test.vdb");
     	conn = server.createConnection("jdbc:teiid:test"); //$NON-NLS-1$ //$NON-NLS-2$		
     }
 	
+	@After public void teardown() throws SQLException {
+		conn.close();
+	}
+	
 	@Test public void testCacheHint() throws Exception {
 		Statement s = conn.createStatement();
 		s.execute("set showplan on");
@@ -56,6 +72,18 @@
 		assertFalse(rs.next());
 	}
 	
+	@Test public void testCacheHintWithMaxRows() throws Exception {
+		Statement s = conn.createStatement();
+		s.setMaxRows(1);
+		ResultSet rs = s.executeQuery("/* cache */ select 1 union all select 2");
+		assertTrue(rs.next());
+		assertFalse(rs.next());
+		s.setMaxRows(2);
+		rs = s.executeQuery("/* cache */ select 1 union all select 2");
+		assertTrue(rs.next());
+		assertTrue(rs.next());
+	}
+	
 	@Test public void testCacheHintTtl() throws Exception {
 		Statement s = conn.createStatement();
 		s.execute("set showplan on");