Author: shawkins Date: 2012-09-17 07:25:25 -0400 (Mon, 17 Sep 2012) New Revision: 4449 Modified: trunk/build/kits/jboss-as7/bin/scripts/teiid-domain-mode-install.cli trunk/build/kits/jboss-as7/docs/teiid/teiid-releasenotes.html trunk/build/kits/jboss-as7/standalone/configuration/standalone-teiid.xml trunk/engine/pom.xml trunk/engine/src/main/java/org/teiid/cache/Cache.java trunk/engine/src/main/java/org/teiid/cache/CacheFactory.java trunk/engine/src/main/java/org/teiid/dqp/internal/process/AccessInfo.java trunk/engine/src/main/java/org/teiid/dqp/internal/process/Request.java trunk/engine/src/main/java/org/teiid/dqp/internal/process/RequestWorkItem.java trunk/engine/src/main/java/org/teiid/dqp/internal/process/SessionAwareCache.java trunk/engine/src/test/java/org/teiid/cache/DefaultCacheFactory.java trunk/engine/src/test/java/org/teiid/dqp/internal/process/TestPreparedPlanCache.java trunk/jboss-integration/src/main/java/org/teiid/jboss/CacheFactoryService.java trunk/jboss-integration/src/main/java/org/teiid/jboss/CacheService.java trunk/jboss-integration/src/main/java/org/teiid/jboss/TeiidAdd.java trunk/runtime/src/main/java/org/teiid/cache/infinispan/InfinispanCache.java trunk/runtime/src/main/java/org/teiid/cache/infinispan/InfinispanCacheFactory.java trunk/runtime/src/main/java/org/teiid/runtime/EmbeddedConfiguration.java trunk/runtime/src/main/resources/infinispan-config.xml trunk/runtime/src/test/java/org/teiid/runtime/TestEmbeddedServer.java trunk/test-integration/common/src/test/java/org/teiid/arquillian/IntegrationTestTransactions.java trunk/test-integration/common/src/test/java/org/teiid/systemmodel/TestReplication.java trunk/test-integration/common/src/test/resources/infinispan-replicated-config.xml Log: TEIID-2086 ensuring that rs cache access is transactionally consistent Modified: trunk/build/kits/jboss-as7/bin/scripts/teiid-domain-mode-install.cli =================================================================== --- trunk/build/kits/jboss-as7/bin/scripts/teiid-domain-mode-install.cli 2012-09-14 17:24:07 UTC (rev 4448) +++ trunk/build/kits/jboss-as7/bin/scripts/teiid-domain-mode-install.cli 2012-09-17 11:25:25 UTC (rev 4449) @@ -10,17 +10,19 @@ /profile=ha/subsystem=infinispan/cache-container=teiid-cache/replicated-cache=resultset-repl:add(mode=SYNC, batching=true) /profile=ha/subsystem=infinispan/cache-container=teiid-cache/replicated-cache=resultset-repl/locking=LOCKING:add(isolation=READ_COMMITTED) /profile=ha/subsystem=infinispan/cache-container=teiid-cache/replicated-cache=resultset-repl/expiration=EXPIRATION:add(lifespan=7200000) -/profile=ha/subsystem=infinispan/cache-container=teiid-cache/replicated-cache=resultset-repl/eviction=EVICTION:add(max-entries=1024, strategy=LRU) +/profile=ha/subsystem=infinispan/cache-container=teiid-cache/replicated-cache=resultset-repl/eviction=EVICTION:add(max-entries=1024, strategy=LIRS) +/profile=ha/subsystem=infinispan/cache-container=teiid-cache/local-cache=resultset/transaction=TRANSACTION:add(mode=NON_XA) /profile=ha/subsystem=infinispan/cache-container=teiid-cache/local-cache=resultset:add(mode=SYNC, batching=true) /profile=ha/subsystem=infinispan/cache-container=teiid-cache/local-cache=resultset/locking=LOCKING:add(isolation=READ_COMMITTED) /profile=ha/subsystem=infinispan/cache-container=teiid-cache/local-cache=resultset/expiration=EXPIRATION:add(lifespan=7200000) -/profile=ha/subsystem=infinispan/cache-container=teiid-cache/local-cache=resultset/eviction=EVICTION:add(max-entries=1024, strategy=LRU) +/profile=ha/subsystem=infinispan/cache-container=teiid-cache/local-cache=resultset/eviction=EVICTION:add(max-entries=1024, strategy=LIRS) +/profile=ha/subsystem=infinispan/cache-container=teiid-cache/local-cache=resultset/transaction=TRANSACTION:add(mode=NON_XA) /profile=ha/subsystem=infinispan/cache-container=teiid-cache/local-cache=preparedplan:add(mode=SYNC, batching=true) /profile=ha/subsystem=infinispan/cache-container=teiid-cache/local-cache=preparedplan/locking=LOCKING:add(isolation=READ_COMMITTED) /profile=ha/subsystem=infinispan/cache-container=teiid-cache/local-cache=preparedplan/expiration=EXPIRATION:add(lifespan=28800) -/profile=ha/subsystem=infinispan/cache-container=teiid-cache/local-cache=preparedplan/eviction=EVICTION:add(max-entries=512, strategy=LRU) +/profile=ha/subsystem=infinispan/cache-container=teiid-cache/local-cache=preparedplan/eviction=EVICTION:add(max-entries=512, strategy=LIRS) /profile=ha/subsystem=security/security-domain=teiid-security:add(cache-type=default) /profile=ha/subsystem=security/security-domain=teiid-security/authentication=classic:add(login-modules=[{"code"=>"org.jboss.security.auth.spi.UsersRolesLoginModule", "flag"=>"required", "module-options"=>[("usersProperties"=>"${jboss.domain.config.dir}/teiid-security-users.properties"), ("rolesProperties"=>"${jboss.domain.config.dir}/teiid-security-roles.properties")]}]) Modified: trunk/build/kits/jboss-as7/docs/teiid/teiid-releasenotes.html =================================================================== --- trunk/build/kits/jboss-as7/docs/teiid/teiid-releasenotes.html 2012-09-14 17:24:07 UTC (rev 4448) +++ trunk/build/kits/jboss-as7/docs/teiid/teiid-releasenotes.html 2012-09-17 11:25:25 UTC (rev 4449) @@ -36,6 +36,7 @@ <li>TEIID-2195 <b>Function Based Indexes</b> - DDL can be used to specify function based indexes. When used on an internal mat view it will create the corresponding function based index automatically. In any other circumstance, the metadata is not currently used. <li>TEIID-2181 <b>System query performance</b> - system queries are now indexed on case-insensitive schema/table name columns. + <li>TEIID-2086 <b>Transactional results caching</b> - the result set cache is now transactional by default. </ul> <h2><a name="Compatibility">Compatibility Issues</a></h2> @@ -163,7 +164,8 @@ <h4>from 8.1</h4> <ul> - <li>The file translator now defaults to exceptionIfFileNotFound=true, you can set the translator property to false to preserve the old behavior of returning null. + <li>The file translator now defaults to exceptionIfFileNotFound=true, you can set the translator property to false to preserve the old behavior of returning null. + <li>TEIID-2086 TEIID-2168 prepared plan and result set caches are now configured as infinispan caches. See the teiid cache container in the configuration. You may also control the transactional aspects of the result set cache on the resultset and resultset-repl caches via the configuration. </ul> <h4>from 8.0</h4> Modified: trunk/build/kits/jboss-as7/standalone/configuration/standalone-teiid.xml =================================================================== --- trunk/build/kits/jboss-as7/standalone/configuration/standalone-teiid.xml 2012-09-14 17:24:07 UTC (rev 4448) +++ trunk/build/kits/jboss-as7/standalone/configuration/standalone-teiid.xml 2012-09-17 11:25:25 UTC (rev 4449) @@ -170,10 +170,12 @@ </cache-container> <cache-container name="teiid" default-cache="resultset"> <local-cache name="resultset"> + <transaction mode="NON_XA"/> <expiration lifespan="7200000" /> <eviction max-entries="1024" strategy="LIRS"/> </local-cache> <local-cache name="resultset-repl"> + <transaction mode="NON_XA"/> <expiration lifespan="7200000" /> <eviction max-entries="1024" strategy="LIRS"/> </local-cache> Modified: trunk/engine/pom.xml =================================================================== --- trunk/engine/pom.xml 2012-09-14 17:24:07 UTC (rev 4448) +++ trunk/engine/pom.xml 2012-09-17 11:25:25 UTC (rev 4449) @@ -98,22 +98,6 @@ <artifactId>xom</artifactId> <version>1.2</version> </dependency> - <dependency> - <groupId>org.infinispan</groupId> - <artifactId>infinispan-core</artifactId> - <scope>test</scope> - </dependency> - <dependency> - <groupId>org.jboss.marshalling</groupId> - <artifactId>jboss-marshalling-river</artifactId> - <scope>test</scope> - </dependency> - - <dependency> - <groupId>org.jboss.marshalling</groupId> - <artifactId>jboss-marshalling</artifactId> - <scope>test</scope> - </dependency> </dependencies> </project> Modified: trunk/engine/src/main/java/org/teiid/cache/Cache.java =================================================================== --- trunk/engine/src/main/java/org/teiid/cache/Cache.java 2012-09-14 17:24:07 UTC (rev 4448) +++ trunk/engine/src/main/java/org/teiid/cache/Cache.java 2012-09-17 11:25:25 UTC (rev 4449) @@ -81,5 +81,12 @@ * Return all the keys * @return */ - Set<K> keys(); + Set<K> keySet(); + + /** + * If the cache is transactional + * @return + */ + boolean isTransactional(); + } Modified: trunk/engine/src/main/java/org/teiid/cache/CacheFactory.java =================================================================== --- trunk/engine/src/main/java/org/teiid/cache/CacheFactory.java 2012-09-14 17:24:07 UTC (rev 4448) +++ trunk/engine/src/main/java/org/teiid/cache/CacheFactory.java 2012-09-17 11:25:25 UTC (rev 4449) @@ -36,9 +36,4 @@ */ void destroy(); - /** - * Return true if replicated caches are created by this factory - * @return - */ - boolean isReplicated(); } Modified: trunk/engine/src/main/java/org/teiid/dqp/internal/process/AccessInfo.java =================================================================== --- trunk/engine/src/main/java/org/teiid/dqp/internal/process/AccessInfo.java 2012-09-14 17:24:07 UTC (rev 4448) +++ trunk/engine/src/main/java/org/teiid/dqp/internal/process/AccessInfo.java 2012-09-17 11:25:25 UTC (rev 4449) @@ -54,18 +54,13 @@ private boolean sensitiveToMetadataChanges = true; private List<List<String>> externalNames; - private transient long creationTime = System.currentTimeMillis(); + private long creationTime = System.currentTimeMillis(); private void writeObject(java.io.ObjectOutputStream out) throws IOException { externalNames = initExternalList(externalNames, objectsAccessed); out.defaultWriteObject(); } - private void readObject(java.io.ObjectInputStream in) throws IOException, ClassNotFoundException { - in.defaultReadObject(); - this.creationTime = System.currentTimeMillis(); - } - public boolean isSensitiveToMetadataChanges() { return sensitiveToMetadataChanges; } Modified: trunk/engine/src/main/java/org/teiid/dqp/internal/process/Request.java =================================================================== --- trunk/engine/src/main/java/org/teiid/dqp/internal/process/Request.java 2012-09-14 17:24:07 UTC (rev 4448) +++ trunk/engine/src/main/java/org/teiid/dqp/internal/process/Request.java 2012-09-17 11:25:25 UTC (rev 4449) @@ -336,14 +336,24 @@ } private void createProcessor() throws TeiidComponentException { + this.context.setTransactionContext(getTransactionContext(true)); + this.processor = new QueryProcessor(processPlan, context, bufferManager, processorDataManager); + this.processor.setContinuous(this.requestMsg.getRequestOptions().isContinuous()); + } + + TransactionContext getTransactionContext(boolean startAutoWrap) throws TeiidComponentException { + if (this.transactionContext != null) { + return this.transactionContext; + } + TransactionContext tc = transactionService.getOrCreateTransactionContext(workContext.getSessionId()); - TransactionContext tc = transactionService.getOrCreateTransactionContext(workContext.getSessionId()); - Assertion.assertTrue(tc.getTransactionType() != TransactionContext.Scope.REQUEST, "Transaction already associated with request."); //$NON-NLS-1$ // If local or global transaction is not started. - if (tc.getTransactionType() == Scope.NONE) { - + if (tc.getTransactionType() == Scope.NONE && !requestMsg.isNoExec()) { + if (!startAutoWrap) { + return null; + } boolean startAutoWrapTxn = false; if(RequestMessage.TXN_WRAP_ON.equals(requestMsg.getTxnAutoWrapMode())){ @@ -365,10 +375,8 @@ tc.setIsolationLevel(requestMsg.getTransactionIsolation()); this.transactionContext = tc; - this.context.setTransactionContext(tc); - this.processor = new QueryProcessor(processPlan, context, bufferManager, processorDataManager); - this.processor.setContinuous(this.requestMsg.getRequestOptions().isContinuous()); - } + return this.transactionContext; + } /** * state side effects: 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-09-14 17:24:07 UTC (rev 4448) +++ trunk/engine/src/main/java/org/teiid/dqp/internal/process/RequestWorkItem.java 2012-09-17 11:25:25 UTC (rev 4449) @@ -510,6 +510,14 @@ cacheId = new CacheID(this.dqpWorkContext, pi, requestMsg.getCommandString()); cachable = cacheId.setParameters(requestMsg.getParameterValues()); if (cachable) { + //allow cache to be transactionally aware + if (rsCache.isTransactional()) { + TransactionContext tc = request.getTransactionContext(false); + if (tc != null && tc.getTransactionType() != Scope.NONE) { + initTransactionState(tc); + resume(); + } + } CachedResults cr = rsCache.get(cacheId); //check that there are enough cached results //TODO: possibly ignore max rows for caching @@ -587,13 +595,7 @@ //This is just a dummy result it will get replaced by collector source resultsBuffer = this.processor.getBufferManager().createTupleBuffer(this.originalCommand.getProjectedSymbols(), this.request.context.getConnectionId(), TupleSourceType.FINAL); } - transactionContext = request.transactionContext; - if (this.transactionContext != null && this.transactionContext.getTransactionType() != Scope.NONE) { - if (this.requestMsg.getRequestOptions().isContinuous()) { - throw new IllegalStateException("Continuous requests are not allowed to be transactional."); //$NON-NLS-1$ - } - this.transactionState = TransactionState.ACTIVE; - } + initTransactionState(request.transactionContext); if (requestMsg.isNoExec()) { doneProducingBatches(); resultsBuffer.close(); @@ -606,6 +608,16 @@ request = null; } + private void initTransactionState(TransactionContext tc) { + transactionContext = tc; + if (this.transactionContext != null && this.transactionContext.getTransactionType() != Scope.NONE) { + if (this.requestMsg.getRequestOptions().isContinuous()) { + throw new IllegalStateException("Continuous requests are not allowed to be transactional."); //$NON-NLS-1$ + } + this.transactionState = TransactionState.ACTIVE; + } + } + private CacheHint getCacheHint() { if (requestMsg.getCommand() != null) { return ((Command)requestMsg.getCommand()).getCacheHint(); @@ -942,7 +954,6 @@ return transactionContext; } - Collection<DataTierTupleSource> getConnectorRequests() { synchronized (this.connectorInfo) { return new ArrayList<DataTierTupleSource>(this.connectorInfo.values()); Modified: trunk/engine/src/main/java/org/teiid/dqp/internal/process/SessionAwareCache.java =================================================================== --- trunk/engine/src/main/java/org/teiid/dqp/internal/process/SessionAwareCache.java 2012-09-14 17:24:07 UTC (rev 4448) +++ trunk/engine/src/main/java/org/teiid/dqp/internal/process/SessionAwareCache.java 2012-09-17 11:25:25 UTC (rev 4449) @@ -48,6 +48,7 @@ * This class is used to cache session aware objects */ public class SessionAwareCache<T> { + public static final String REPL = "-repl"; //$NON-NLS-1$ public static final int DEFAULT_MAX_SIZE_TOTAL = 512; public enum Type { RESULTSET, @@ -75,7 +76,7 @@ this.distributedCache = localCache; } else { - this.distributedCache = cacheFactory.get(cacheName+"-repl"); //$NON-NLS-1$ + this.distributedCache = cacheFactory.get(cacheName+REPL); if (this.distributedCache == null && this.localCache != null) { this.distributedCache = this.localCache; } @@ -204,7 +205,7 @@ } private void clearCache(Cache<CacheID, T> cache, String vdbName, int version) { - Set<CacheID> keys = cache.keys(); + Set<CacheID> keys = cache.keySet(); VDBKey vdbKey = new VDBKey(vdbName, version); for (CacheID key:keys) { if (key.vdbInfo.equals(vdbKey)) { @@ -323,4 +324,8 @@ public static boolean isResultsetCache(String cacheType) { return (Admin.Cache.valueOf(cacheType) == Admin.Cache.QUERY_SERVICE_RESULT_SET_CACHE); } + + public boolean isTransactional() { + return this.localCache.isTransactional() || this.distributedCache.isTransactional(); + } } Modified: trunk/engine/src/test/java/org/teiid/cache/DefaultCacheFactory.java =================================================================== --- trunk/engine/src/test/java/org/teiid/cache/DefaultCacheFactory.java 2012-09-14 17:24:07 UTC (rev 4448) +++ trunk/engine/src/test/java/org/teiid/cache/DefaultCacheFactory.java 2012-09-17 11:25:25 UTC (rev 4449) @@ -22,19 +22,15 @@ package org.teiid.cache; import java.io.Serializable; -import java.util.Set; -import java.util.concurrent.TimeUnit; -import org.infinispan.configuration.cache.Configuration; -import org.infinispan.configuration.cache.ConfigurationBuilder; -import org.infinispan.eviction.EvictionStrategy; -import org.infinispan.manager.DefaultCacheManager; -import org.infinispan.manager.EmbeddedCacheManager; import org.teiid.cache.CacheConfiguration.Policy; import org.teiid.core.TeiidRuntimeException; +import org.teiid.core.util.LRUCache; import org.teiid.query.QueryPlugin; -@SuppressWarnings("nls") +/** + * Provides a non-thread safe simple map backed cache suitable for testing + */ public class DefaultCacheFactory implements CacheFactory, Serializable { private static final long serialVersionUID = -5541424157695857527L; private static CacheConfiguration DEFAULT = new CacheConfiguration(Policy.LRU, 60*60, 100, "default"); // 1 hours with 100 nodes. //$NON-NLS-1$ @@ -42,98 +38,48 @@ public static DefaultCacheFactory INSTANCE = new DefaultCacheFactory(DEFAULT); private volatile boolean destroyed = false; - EmbeddedCacheManager manager; - + private CacheConfiguration config; public DefaultCacheFactory(CacheConfiguration config) { - Configuration cacheConfig = new ConfigurationBuilder().eviction() - .strategy(config.getPolicy()==Policy.LRU?EvictionStrategy.LRU:EvictionStrategy.NONE) - .maxEntries(config.getMaxEntries()) - .expiration().lifespan(config.getMaxAgeInSeconds()*1000) - .wakeUpInterval(200) - .build(); - this.manager = new DefaultCacheManager(cacheConfig); - this.manager.start(); - this.manager.defineConfiguration("resultset", cacheConfig); - this.manager.defineConfiguration("resultset-repl", cacheConfig); - this.manager.defineConfiguration("preparedplan", cacheConfig); - this.manager.getCache("resultset"); - this.manager.getCache("preparedplan"); - this.manager.getCache("resultset-repl"); + this.config = config; } @Override public void destroy() { this.destroyed = true; - if (this.manager != null) { - this.manager.stop(); - } } @Override public <K, V> Cache<K, V> get(String cacheName) { if (!destroyed) { - manager.getCache(cacheName).clear(); - return new IspnCache(manager.getCache(cacheName), cacheName, getClass().getClassLoader()); + return new MockCache<K, V>(cacheName, config.getMaxEntries()); } throw new TeiidRuntimeException(QueryPlugin.Event.TEIID30562, QueryPlugin.Util.gs(QueryPlugin.Event.TEIID30562)); } - @Override - public boolean isReplicated() { - return false; - } - - - private static class IspnCache<K, V> implements Cache<K, V> { + @SuppressWarnings("serial") + private static class MockCache<K, V> extends LRUCache<K, V> implements Cache<K, V> { - protected org.infinispan.AdvancedCache<K, V> cacheStore; - private final String name; - private ClassLoader classloader; + private String name; - public IspnCache(org.infinispan.Cache<K, V> cacheStore, String cacheName, ClassLoader classloader) { - assert(cacheStore != null); - this.cacheStore = cacheStore.getAdvancedCache(); + public MockCache(String cacheName, int maxSize) { + super(maxSize<0?Integer.MAX_VALUE:maxSize); this.name = cacheName; - this.classloader = classloader; } @Override - public V get(K key) { - return this.cacheStore.with(this.classloader).get(key); - } - - @Override public V put(K key, V value, Long ttl) { - if (ttl != null) { - return this.cacheStore.with(this.classloader).put(key, value, ttl, TimeUnit.MILLISECONDS); - } - return this.cacheStore.with(this.classloader).put(key, value); + return put(key, value); } @Override - public V remove(K key) { - return this.cacheStore.with(this.classloader).remove(key); - } - - @Override - public int size() { - return this.cacheStore.with(this.classloader).size(); - } - - @Override - public void clear() { - this.cacheStore.with(this.classloader).clear(); - } - - @Override public String getName() { return this.name; } @Override - public Set<K> keys() { - return this.cacheStore.with(this.classloader).keySet(); + public boolean isTransactional() { + return false; } } } Modified: trunk/engine/src/test/java/org/teiid/dqp/internal/process/TestPreparedPlanCache.java =================================================================== --- trunk/engine/src/test/java/org/teiid/dqp/internal/process/TestPreparedPlanCache.java 2012-09-14 17:24:07 UTC (rev 4448) +++ trunk/engine/src/test/java/org/teiid/dqp/internal/process/TestPreparedPlanCache.java 2012-09-17 11:25:25 UTC (rev 4449) @@ -30,7 +30,6 @@ import org.junit.Test; import org.teiid.api.exception.query.QueryParserException; import org.teiid.cache.CacheConfiguration; -import org.teiid.cache.CacheConfiguration.Policy; import org.teiid.cache.DefaultCacheFactory; import org.teiid.dqp.internal.process.SessionAwareCache.CacheID; import org.teiid.metadata.FunctionMethod.Determinism; @@ -101,56 +100,11 @@ assertNull("Unable to get prepared plan from cache for token2", cache.get(new CacheID(token2, pi, EXAMPLE_QUERY + 1))); //$NON-NLS-1$ } - @Test public void testMaxSize() throws Exception { - CacheConfiguration config = new CacheConfiguration(); - config.setType(Policy.LRU.name()); - config.setMaxEntries(100); - config.setMaxAgeInSeconds(60); - SessionAwareCache<PreparedPlan> cache = new SessionAwareCache<PreparedPlan>("preparedplan", new DefaultCacheFactory(config), SessionAwareCache.Type.PREPAREDPLAN, 0); - - helpPutPreparedPlans(cache, token, 0, 101); - assertTrue(cache.getTotalCacheEntries() <= 100); - - // find a entry that has not been evicted. - int i = 0; - while (true) { - PreparedPlan plan = cache.get(new CacheID(token, pi, EXAMPLE_QUERY + i)); - if (plan != null) { - break; - } - i++; - if (i > 100) break; - } - assertNotNull(cache.get(new CacheID(token, pi, EXAMPLE_QUERY + i))); - - helpPutPreparedPlans(cache, token, 102, 50); - - //"sql12" should still be there based on lru policy - assertNotNull(cache.get(new CacheID(token, pi, EXAMPLE_QUERY + i))); - - helpPutPreparedPlans(cache, token2, 0, 121); - helpPutPreparedPlans(cache, token, 0, 50); - assertTrue(cache.getTotalCacheEntries() <= 100); - } - - @Test public void testZeroSizeCache() throws Exception { - CacheConfiguration config = new CacheConfiguration(); - config.setMaxEntries(0); - config.setType(Policy.LRU.name()); - - try { - // Create with 0 size cache - SessionAwareCache<PreparedPlan> cache = new SessionAwareCache<PreparedPlan>("preparedplan", new DefaultCacheFactory(config), SessionAwareCache.Type.PREPAREDPLAN, 0); - fail("should have failed to create zero sized cache store"); - } catch (Exception e) { - } - } - // set init size to negative number, which should default to max @Test public void testNegativeSizeCacheUsesDefault() { CacheConfiguration config = new CacheConfiguration(); config.setMaxEntries(-1); - SessionAwareCache<PreparedPlan> negativeSizedCache = new SessionAwareCache<PreparedPlan>("preparedplan", new DefaultCacheFactory(config), SessionAwareCache.Type.PREPAREDPLAN, 0); + new SessionAwareCache<PreparedPlan>("preparedplan", new DefaultCacheFactory(config), SessionAwareCache.Type.PREPAREDPLAN, 0); // -1 means unlimited in the infinispan } Modified: trunk/jboss-integration/src/main/java/org/teiid/jboss/CacheFactoryService.java =================================================================== --- trunk/jboss-integration/src/main/java/org/teiid/jboss/CacheFactoryService.java 2012-09-14 17:24:07 UTC (rev 4448) +++ trunk/jboss-integration/src/main/java/org/teiid/jboss/CacheFactoryService.java 2012-09-17 11:25:25 UTC (rev 4449) @@ -21,7 +21,7 @@ */ package org.teiid.jboss; -import org.infinispan.manager.CacheContainer; +import org.infinispan.manager.EmbeddedCacheManager; import org.jboss.modules.Module; import org.jboss.msc.service.Service; import org.jboss.msc.service.StartContext; @@ -32,12 +32,12 @@ import org.teiid.cache.infinispan.InfinispanCacheFactory; class CacheFactoryService implements Service<CacheFactory> { - protected InjectedValue<CacheContainer> cacheContainerInjector = new InjectedValue<CacheContainer>(); + protected InjectedValue<EmbeddedCacheManager> cacheContainerInjector = new InjectedValue<EmbeddedCacheManager>(); private CacheFactory cacheFactory; @Override public void start(StartContext context) throws StartException { - CacheContainer cc = cacheContainerInjector.getValue(); + EmbeddedCacheManager cc = cacheContainerInjector.getValue(); if (cc != null) { this.cacheFactory = new InfinispanCacheFactory(cc, Module.getCallerModule().getClassLoader()); } Modified: trunk/jboss-integration/src/main/java/org/teiid/jboss/CacheService.java =================================================================== --- trunk/jboss-integration/src/main/java/org/teiid/jboss/CacheService.java 2012-09-14 17:24:07 UTC (rev 4448) +++ trunk/jboss-integration/src/main/java/org/teiid/jboss/CacheService.java 2012-09-17 11:25:25 UTC (rev 4449) @@ -29,6 +29,7 @@ import org.teiid.cache.CacheFactory; import org.teiid.common.buffer.TupleBufferCache; import org.teiid.dqp.internal.process.SessionAwareCache; +import org.teiid.dqp.internal.process.SessionAwareCache.Type; class CacheService<T> implements Service<SessionAwareCache<T>> { @@ -49,7 +50,9 @@ @Override public void start(StartContext context) throws StartException { this.cache = new SessionAwareCache<T>(this.cacheName, cacheFactoryInjector.getValue(), this.type, this.maxStaleness); - this.cache.setTupleBufferCache(this.tupleBufferCacheInjector.getValue()); + if (type == Type.RESULTSET) { + this.cache.setTupleBufferCache(this.tupleBufferCacheInjector.getValue()); + } } @Override Modified: trunk/jboss-integration/src/main/java/org/teiid/jboss/TeiidAdd.java =================================================================== --- trunk/jboss-integration/src/main/java/org/teiid/jboss/TeiidAdd.java 2012-09-14 17:24:07 UTC (rev 4448) +++ trunk/jboss-integration/src/main/java/org/teiid/jboss/TeiidAdd.java 2012-09-17 11:25:25 UTC (rev 4449) @@ -22,10 +22,7 @@ package org.teiid.jboss; -import static org.jboss.as.controller.descriptions.ModelDescriptionConstants.ADD; -import static org.jboss.as.controller.descriptions.ModelDescriptionConstants.DESCRIPTION; -import static org.jboss.as.controller.descriptions.ModelDescriptionConstants.OPERATION_NAME; -import static org.jboss.as.controller.descriptions.ModelDescriptionConstants.REQUEST_PROPERTIES; +import static org.jboss.as.controller.descriptions.ModelDescriptionConstants.*; import java.util.Iterator; import java.util.List; @@ -38,15 +35,15 @@ import javax.resource.spi.work.WorkManager; import javax.transaction.TransactionManager; -import org.infinispan.manager.CacheContainer; +import org.infinispan.manager.EmbeddedCacheManager; import org.jboss.as.clustering.jgroups.ChannelFactory; import org.jboss.as.controller.AbstractAddStepHandler; import org.jboss.as.controller.OperationContext; import org.jboss.as.controller.OperationFailedException; import org.jboss.as.controller.ServiceVerificationHandler; import org.jboss.as.controller.descriptions.DescriptionProvider; -import org.jboss.as.controller.registry.AttributeAccess.Storage; import org.jboss.as.controller.registry.ManagementResourceRegistration; +import org.jboss.as.controller.registry.AttributeAccess.Storage; import org.jboss.as.controller.services.path.RelativePathService; import org.jboss.as.naming.ManagedReferenceFactory; import org.jboss.as.naming.ServiceBasedNamingStore; @@ -61,12 +58,12 @@ import org.jboss.modules.ModuleIdentifier; import org.jboss.modules.ModuleLoadException; import org.jboss.msc.service.ServiceBuilder; -import org.jboss.msc.service.ServiceBuilder.DependencyType; import org.jboss.msc.service.ServiceContainer; import org.jboss.msc.service.ServiceController; import org.jboss.msc.service.ServiceName; import org.jboss.msc.service.ServiceTarget; import org.jboss.msc.service.ValueService; +import org.jboss.msc.service.ServiceBuilder.DependencyType; import org.jboss.msc.value.InjectedValue; import org.teiid.PolicyDecider; import org.teiid.cache.CacheFactory; @@ -321,7 +318,7 @@ ServiceBuilder<CacheFactory> cacheFactoryBuilder = target.addService(cfName, cfs); String ispnName = Element.RSC_CONTAINER_NAME_ELEMENT.asString(operation); - cacheFactoryBuilder.addDependency(ServiceName.JBOSS.append("infinispan", ispnName), CacheContainer.class, cfs.cacheContainerInjector); //$NON-NLS-1$ + cacheFactoryBuilder.addDependency(ServiceName.JBOSS.append("infinispan", ispnName), EmbeddedCacheManager.class, cfs.cacheContainerInjector); //$NON-NLS-1$ newControllers.add(cacheFactoryBuilder.install()); int maxStaleness = 60; @@ -333,6 +330,8 @@ ServiceBuilder<SessionAwareCache<CachedResults>> resultsCacheBuilder = target.addService(TeiidServiceNames.CACHE_RESULTSET, resultSetService); resultsCacheBuilder.addDependency(TeiidServiceNames.TUPLE_BUFFER, TupleBufferCache.class, resultSetService.tupleBufferCacheInjector); resultsCacheBuilder.addDependency(cfName, CacheFactory.class, resultSetService.cacheFactoryInjector); + resultsCacheBuilder.addDependency(ServiceName.JBOSS.append("infinispan", ispnName, cacheName)); //$NON-NLS-1$ + resultsCacheBuilder.addDependency(ServiceName.JBOSS.append("infinispan", ispnName, cacheName + SessionAwareCache.REPL)); //$NON-NLS-1$ newControllers.add(resultsCacheBuilder.install()); } @@ -357,13 +356,13 @@ ServiceBuilder<CacheFactory> cacheFactoryBuilder = target.addService(cfName, cfs); String ispnName = Element.PPC_CONTAINER_NAME_ELEMENT.asString(operation); - cacheFactoryBuilder.addDependency(ServiceName.JBOSS.append("infinispan", ispnName), CacheContainer.class, cfs.cacheContainerInjector); //$NON-NLS-1$ - newControllers.add(cacheFactoryBuilder.install()); + cacheFactoryBuilder.addDependency(ServiceName.JBOSS.append("infinispan", ispnName), EmbeddedCacheManager.class, cfs.cacheContainerInjector); //$NON-NLS-1$ + newControllers.add(cacheFactoryBuilder.install()); CacheService<PreparedPlan> preparedPlanService = new CacheService<PreparedPlan>(cacheName, SessionAwareCache.Type.PREPAREDPLAN, 0); ServiceBuilder<SessionAwareCache<PreparedPlan>> preparedPlanCacheBuilder = target.addService(TeiidServiceNames.CACHE_PREPAREDPLAN, preparedPlanService); - preparedPlanCacheBuilder.addDependency(TeiidServiceNames.TUPLE_BUFFER, TupleBufferCache.class, preparedPlanService.tupleBufferCacheInjector); preparedPlanCacheBuilder.addDependency(cfName, CacheFactory.class, preparedPlanService.cacheFactoryInjector); + preparedPlanCacheBuilder.addDependency(ServiceName.JBOSS.append("infinispan", ispnName, cacheName)); //$NON-NLS-1$ newControllers.add(preparedPlanCacheBuilder.install()); } Modified: trunk/runtime/src/main/java/org/teiid/cache/infinispan/InfinispanCache.java =================================================================== --- trunk/runtime/src/main/java/org/teiid/cache/infinispan/InfinispanCache.java 2012-09-14 17:24:07 UTC (rev 4448) +++ trunk/runtime/src/main/java/org/teiid/cache/infinispan/InfinispanCache.java 2012-09-17 11:25:25 UTC (rev 4449) @@ -25,7 +25,10 @@ import java.util.Set; import java.util.concurrent.TimeUnit; +import org.infinispan.transaction.TransactionMode; import org.teiid.cache.Cache; +import org.teiid.logging.LogConstants; +import org.teiid.logging.LogManager; /** @@ -36,15 +39,24 @@ protected org.infinispan.AdvancedCache<K, V> cacheStore; private final String name; private ClassLoader classloader; + private boolean transactional; public InfinispanCache(org.infinispan.Cache<K, V> cacheStore, String cacheName, ClassLoader classloader) { assert(cacheStore != null); this.cacheStore = cacheStore.getAdvancedCache(); + TransactionMode transactionMode = this.cacheStore.getCacheConfiguration().transaction().transactionMode(); + this.transactional = transactionMode == TransactionMode.TRANSACTIONAL; + LogManager.logDetail(LogConstants.CTX_RUNTIME, "Added", transactionMode, "infinispan cache", cacheName); //$NON-NLS-1$ //$NON-NLS-2$ this.name = cacheName; this.classloader = classloader; } @Override + public boolean isTransactional() { + return transactional; + } + + @Override public V get(K key) { return this.cacheStore.with(this.classloader).get(key); } @@ -82,7 +94,7 @@ } @Override - public Set<K> keys() { + public Set<K> keySet() { return this.cacheStore.with(this.classloader).keySet(); } } Modified: trunk/runtime/src/main/java/org/teiid/cache/infinispan/InfinispanCacheFactory.java =================================================================== --- trunk/runtime/src/main/java/org/teiid/cache/infinispan/InfinispanCacheFactory.java 2012-09-14 17:24:07 UTC (rev 4448) +++ trunk/runtime/src/main/java/org/teiid/cache/infinispan/InfinispanCacheFactory.java 2012-09-17 11:25:25 UTC (rev 4449) @@ -24,30 +24,30 @@ import java.io.Serializable; -import org.infinispan.manager.CacheContainer; +import org.infinispan.manager.EmbeddedCacheManager; import org.teiid.cache.Cache; import org.teiid.cache.CacheFactory; import org.teiid.core.TeiidRuntimeException; import org.teiid.runtime.RuntimePlugin; - public class InfinispanCacheFactory implements CacheFactory, Serializable{ private static final long serialVersionUID = -2767452034178675653L; - private transient org.infinispan.manager.CacheContainer cacheStore; + private transient EmbeddedCacheManager cacheStore; private volatile boolean destroyed = false; private ClassLoader classLoader; - - public InfinispanCacheFactory(CacheContainer cm, ClassLoader classLoader) { + public InfinispanCacheFactory(EmbeddedCacheManager cm, ClassLoader classLoader) { this.cacheStore = cm; this.classLoader = classLoader; } - public Cache get(String cacheName) { + @SuppressWarnings("unchecked") + @Override + public <K, V> Cache<K, V> get(String cacheName) { if (!destroyed) { - org.infinispan.Cache cache = this.cacheStore.getCache(cacheName); + org.infinispan.Cache cache = this.cacheStore.getCache(cacheName, false); if (cache != null) { - return new InfinispanCache(cache, cacheName, this.classLoader); + return new InfinispanCache<K, V>(cache, cacheName, this.classLoader); } return null; } @@ -62,8 +62,4 @@ destroy(); } - @Override - public boolean isReplicated() { - return true; - } } Modified: trunk/runtime/src/main/java/org/teiid/runtime/EmbeddedConfiguration.java =================================================================== --- trunk/runtime/src/main/java/org/teiid/runtime/EmbeddedConfiguration.java 2012-09-14 17:24:07 UTC (rev 4448) +++ trunk/runtime/src/main/java/org/teiid/runtime/EmbeddedConfiguration.java 2012-09-17 11:25:25 UTC (rev 4449) @@ -49,6 +49,7 @@ private String bufferDirectory; private CacheFactory cacheFactory; private int maxResultSetCacheStaleness = 60; + private String infinispanConfigFile = "infinispan-config.xml"; //$NON-NLS-1$ public SecurityHelper getSecurityHelper() { return securityHelper; @@ -121,10 +122,19 @@ return this.bufferDirectory; } + public String getInfinispanConfigFile() { + return infinispanConfigFile; + } + + public void setInfinispanConfigFile(String infinispanConfigFile) { + this.infinispanConfigFile = infinispanConfigFile; + } + public CacheFactory getCacheFactory() { if (this.cacheFactory == null) { try { - DefaultCacheManager manager = new DefaultCacheManager("infinispan-config.xml"); //$NON-NLS-1$ + DefaultCacheManager manager = new DefaultCacheManager(this.infinispanConfigFile, true); + manager.startCaches(manager.getCacheNames().toArray(new String[manager.getCacheNames().size()])); this.cacheFactory = new InfinispanCacheFactory(manager, this.getClass().getClassLoader()); } catch (IOException e) { throw new TeiidRuntimeException(RuntimePlugin.Event.TEIID40100, e); Modified: trunk/runtime/src/main/resources/infinispan-config.xml =================================================================== --- trunk/runtime/src/main/resources/infinispan-config.xml 2012-09-14 17:24:07 UTC (rev 4448) +++ trunk/runtime/src/main/resources/infinispan-config.xml 2012-09-17 11:25:25 UTC (rev 4449) @@ -30,12 +30,14 @@ <global/> <namedCache name="resultset"> + <transaction transactionMode="TRANSACTIONAL"/> <eviction maxEntries="1024" strategy="LIRS" /> <expiration lifespan="7200000" /> <clustering mode="local"/> </namedCache> <namedCache name="resultset-repl"> + <transaction transactionMode="TRANSACTIONAL"/> <eviction maxEntries="1024" strategy="LIRS" /> <expiration lifespan="7200000" /> <clustering mode="local"/> Modified: trunk/runtime/src/test/java/org/teiid/runtime/TestEmbeddedServer.java =================================================================== --- trunk/runtime/src/test/java/org/teiid/runtime/TestEmbeddedServer.java 2012-09-14 17:24:07 UTC (rev 4448) +++ trunk/runtime/src/test/java/org/teiid/runtime/TestEmbeddedServer.java 2012-09-17 11:25:25 UTC (rev 4449) @@ -360,6 +360,11 @@ assertEquals(1, tm.txnHistory.size()); txn = tm.txnHistory.remove(0); Mockito.verify(txn).commit(); + + s.execute("set autoCommitTxn on"); + s.execute("set noexec on"); + s.execute("select 1"); + assertFalse(s.getResultSet().next()); } } Modified: trunk/test-integration/common/src/test/java/org/teiid/arquillian/IntegrationTestTransactions.java =================================================================== --- trunk/test-integration/common/src/test/java/org/teiid/arquillian/IntegrationTestTransactions.java 2012-09-14 17:24:07 UTC (rev 4448) +++ trunk/test-integration/common/src/test/java/org/teiid/arquillian/IntegrationTestTransactions.java 2012-09-17 11:25:25 UTC (rev 4449) @@ -74,6 +74,34 @@ execute("rollback"); execute("select * from temp"); assertRowCount(0); + + execute("select rand(1)"); //TODO - I think our rand function doesn't make sense + this.internalConnection.setAutoCommit(false); + + execute("/*+ cache */ select rand()"); + internalResultSet.next(); + double d = internalResultSet.getDouble(1); + + execute("select rand(2)"); + execute("/*+ cache */ select rand()"); + internalResultSet.next(); + double d1 = internalResultSet.getDouble(1); + assertEquals("Expected same in the txn", d, d1, 0); + + this.internalConnection.rollback(); + this.internalConnection.setAutoCommit(true); + + execute("select rand(3)"); + execute("/*+ cache */ select rand()"); + internalResultSet.next(); + double d2 = internalResultSet.getDouble(1); + assertTrue("Expected different after rollback", d != d2); + + execute("select rand(4)"); + execute("/*+ cache */ select rand()"); + internalResultSet.next(); + d = internalResultSet.getDouble(1); + assertEquals("Expected same after autoCommit", d, d2, 0); } } Modified: trunk/test-integration/common/src/test/java/org/teiid/systemmodel/TestReplication.java =================================================================== --- trunk/test-integration/common/src/test/java/org/teiid/systemmodel/TestReplication.java 2012-09-14 17:24:07 UTC (rev 4448) +++ trunk/test-integration/common/src/test/java/org/teiid/systemmodel/TestReplication.java 2012-09-17 11:25:25 UTC (rev 4449) @@ -31,14 +31,12 @@ import java.util.Collection; import java.util.HashMap; -import org.infinispan.manager.DefaultCacheManager; import org.jboss.as.clustering.jgroups.ChannelFactory; import org.jboss.as.server.ServerEnvironment; import org.jgroups.Channel; import org.jgroups.JChannel; import org.junit.BeforeClass; import org.junit.Test; -import org.teiid.cache.infinispan.InfinispanCacheFactory; import org.teiid.core.types.DataTypeManager; import org.teiid.core.util.UnitTestUtil; import org.teiid.jdbc.FakeServer; @@ -146,8 +144,7 @@ EmbeddedConfiguration config = new EmbeddedConfiguration(); config.setObjectReplicator(jor); - DefaultCacheManager manager = new DefaultCacheManager(UnitTestUtil.getTestDataPath()+"/infinispan-replicated-config.xml"); - config.setCacheFactory(new InfinispanCacheFactory(manager, this.getClass().getClassLoader())); + config.setInfinispanConfigFile(UnitTestUtil.getTestDataPath()+"/infinispan-replicated-config.xml"); server.start(config, true); HashMap<String, Collection<FunctionMethod>> udfs = new HashMap<String, Collection<FunctionMethod>>(); Modified: trunk/test-integration/common/src/test/resources/infinispan-replicated-config.xml =================================================================== --- trunk/test-integration/common/src/test/resources/infinispan-replicated-config.xml 2012-09-14 17:24:07 UTC (rev 4448) +++ trunk/test-integration/common/src/test/resources/infinispan-replicated-config.xml 2012-09-17 11:25:25 UTC (rev 4449) @@ -11,15 +11,17 @@ </global> <namedCache name="resultset"> + <transaction transactionMode="TRANSACTIONAL"/> <eviction maxEntries="1024" strategy="LIRS" /> <expiration lifespan="7200000"/> <clustering mode="local"/> </namedCache> <namedCache name="resultset-repl"> + <transaction transactionMode="TRANSACTIONAL"/> <eviction maxEntries="1024" strategy="LIRS" /> <expiration lifespan="7200000"/> - <clustering mode="replication"> + <clustering mode="repl"> <sync/> </clustering> </namedCache>