Author: rareddy Date: 2010-07-28 14:04:14 -0400 (Wed, 28 Jul 2010) New Revision: 2386 Added: trunk/cache-jbosscache/src/main/java/org/teiid/cache/jboss/ClusterableCacheFactory.java trunk/engine/src/main/java/org/teiid/cache/DefaultCache.java Modified: trunk/build/kits/jboss-container/deploy/teiid/teiid-jboss-beans.xml trunk/cache-jbosscache/pom.xml trunk/cache-jbosscache/src/main/java/org/teiid/cache/jboss/JBossCacheFactory.java trunk/engine/src/main/java/org/teiid/cache/Cache.java trunk/engine/src/main/java/org/teiid/cache/DefaultCacheFactory.java trunk/engine/src/main/java/org/teiid/dqp/internal/process/CachedResults.java trunk/engine/src/main/java/org/teiid/dqp/internal/process/DQPCore.java trunk/engine/src/main/java/org/teiid/dqp/internal/process/PreparedStatementRequest.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/main/java/org/teiid/query/function/FunctionDescriptor.java trunk/engine/src/main/java/org/teiid/query/function/metadata/FunctionMethod.java trunk/engine/src/main/java/org/teiid/query/function/source/SystemSource.java trunk/engine/src/main/java/org/teiid/query/parser/ParseInfo.java trunk/engine/src/main/java/org/teiid/query/util/CommandContext.java trunk/engine/src/test/java/org/teiid/dqp/internal/process/TestPreparedPlanCache.java trunk/jboss-integration/pom.xml trunk/jboss-integration/src/main/java/org/teiid/jboss/deployers/RuntimeEngineDeployer.java trunk/runtime/src/main/java/org/teiid/transport/ODBCSocketListener.java trunk/runtime/src/main/java/org/teiid/transport/SocketListener.java Log: TEIID-1104: Added "ports-*" flag work in the clustered mode, which automatically offsets the ports in Teiid. Also added capability to hook into JBoss Cache in clustered mode for Teiid caching purposes. However, the Teiid does not make use of the distributed cache yet for its ResultSet Cache or metadata cache. Which is coming in next installment. Modified: trunk/build/kits/jboss-container/deploy/teiid/teiid-jboss-beans.xml =================================================================== --- trunk/build/kits/jboss-container/deploy/teiid/teiid-jboss-beans.xml 2010-07-28 17:06:09 UTC (rev 2385) +++ trunk/build/kits/jboss-container/deploy/teiid/teiid-jboss-beans.xml 2010-07-28 18:04:14 UTC (rev 2386) @@ -44,6 +44,10 @@ <property name="maxOpenFiles">64</property> </bean> + <bean name="CacheFactory" class="org.teiid.cache.jboss.ClusterableCacheFactory"> + <property name="clusteredCacheName">mvcc-shared</property> + </bean> + <bean name="RuntimeEngineDeployer" class="org.teiid.jboss.deployers.RuntimeEngineDeployer"> <property name="jndiName">teiid/engine-deployer</property> <property name="profileService"><inject bean="ProfileService"/></property> @@ -58,7 +62,9 @@ <property name="connectorManagerRepository"><inject bean="ConnectorManagerRepository"/></property> <property name="securityHelper"><inject bean="SecurityHelper"/></property> <property name="VDBRepository"><inject bean="VDBRepository"/></property> + <property name="cacheFactory"><inject bean="CacheFactory"/></property> + <!-- Process pool maximum thread count. (default 64) --> <property name="maxThreads">64</property> <!-- Max active plans (default 20). Increase this value on highly concurrent systems - but ensure that the underlying pools can handle the increased load without timeouts. --> Modified: trunk/cache-jbosscache/pom.xml =================================================================== --- trunk/cache-jbosscache/pom.xml 2010-07-28 17:06:09 UTC (rev 2385) +++ trunk/cache-jbosscache/pom.xml 2010-07-28 18:04:14 UTC (rev 2386) @@ -24,6 +24,7 @@ <dependency> <groupId>org.jboss.cache</groupId> <artifactId>jbosscache-core</artifactId> + <scope>provided</scope> </dependency> </dependencies> </project> \ No newline at end of file Added: trunk/cache-jbosscache/src/main/java/org/teiid/cache/jboss/ClusterableCacheFactory.java =================================================================== --- trunk/cache-jbosscache/src/main/java/org/teiid/cache/jboss/ClusterableCacheFactory.java (rev 0) +++ trunk/cache-jbosscache/src/main/java/org/teiid/cache/jboss/ClusterableCacheFactory.java 2010-07-28 18:04:14 UTC (rev 2386) @@ -0,0 +1,77 @@ +/* + * JBoss, Home of Professional Open Source. + * See the COPYRIGHT.txt file distributed with this work for information + * regarding copyright ownership. Some portions may be licensed + * to Red Hat, Inc. under one or more contributor license agreements. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA + * 02110-1301 USA. + */ +package org.teiid.cache.jboss; + +import java.io.Serializable; + +import javax.naming.Context; +import javax.naming.InitialContext; +import javax.naming.NamingException; + +import org.teiid.cache.Cache; +import org.teiid.cache.CacheConfiguration; +import org.teiid.cache.CacheFactory; +import org.teiid.cache.DefaultCacheFactory; +import org.teiid.cache.Cache.Type; +import org.teiid.core.TeiidRuntimeException; + +public class ClusterableCacheFactory implements CacheFactory, Serializable { + private static final long serialVersionUID = -1992994494154581234L; + private CacheFactory delegate; + private String cacheName; + + @Override + public <K, V> Cache<K, V> get(Type type, CacheConfiguration config) { + if (this.delegate == null) { + Object cacheManager = getClusteredCache(); + if (cacheManager == null) { + this.delegate = new DefaultCacheFactory(); + } + else { + try { + this.delegate = new JBossCacheFactory(this.cacheName, cacheManager); + } catch (Exception e) { + throw new TeiidRuntimeException("Failed to obtain the clusted cache"); //$NON-NLS-1$ + } + } + } + return delegate.get(type, config); + } + + public void setClusteredCacheName(String name) { + this.cacheName = name; + } + + @Override + public void destroy() { + this.delegate.destroy(); + } + + private Object getClusteredCache() { + try { + Context ctx = new InitialContext(); + return ctx.lookup("java:CacheManager"); //$NON-NLS-1$ + } catch (NamingException e) { + return null; + } + } +} Property changes on: trunk/cache-jbosscache/src/main/java/org/teiid/cache/jboss/ClusterableCacheFactory.java ___________________________________________________________________ Name: svn:mime-type + text/plain Modified: trunk/cache-jbosscache/src/main/java/org/teiid/cache/jboss/JBossCacheFactory.java =================================================================== --- trunk/cache-jbosscache/src/main/java/org/teiid/cache/jboss/JBossCacheFactory.java 2010-07-28 17:06:09 UTC (rev 2385) +++ trunk/cache-jbosscache/src/main/java/org/teiid/cache/jboss/JBossCacheFactory.java 2010-07-28 18:04:14 UTC (rev 2386) @@ -24,12 +24,7 @@ import java.io.Serializable; -import javax.management.MBeanServer; -import javax.management.MBeanServerFactory; -import javax.management.MBeanServerInvocationHandler; -import javax.management.MalformedObjectNameException; -import javax.management.ObjectName; - +import org.jboss.cache.CacheManager; import org.jboss.cache.Fqn; import org.jboss.cache.Node; import org.jboss.cache.Region; @@ -38,7 +33,6 @@ import org.jboss.cache.eviction.FIFOAlgorithmConfig; import org.jboss.cache.eviction.LFUAlgorithmConfig; import org.jboss.cache.eviction.LRUAlgorithmConfig; -import org.jboss.cache.jmx.CacheJmxWrapperMBean; import org.teiid.cache.Cache; import org.teiid.cache.CacheConfiguration; import org.teiid.cache.CacheFactory; @@ -48,19 +42,14 @@ public class JBossCacheFactory implements CacheFactory, Serializable{ + private static final long serialVersionUID = -2767452034178675653L; private transient org.jboss.cache.Cache cacheStore; private volatile boolean destroyed = false; - public void setCacheName(String jmxName) { - try { - MBeanServer server = MBeanServerFactory.findMBeanServer(null).get(0); - ObjectName on = new ObjectName(jmxName); - CacheJmxWrapperMBean cacheWrapper = MBeanServerInvocationHandler.newProxyInstance(server, on, CacheJmxWrapperMBean.class, false); - this.cacheStore = cacheWrapper.getCache(); - } catch (MalformedObjectNameException e) { - throw new TeiidRuntimeException(e); - } + public JBossCacheFactory(String name, Object cm) throws Exception { + CacheManager cachemanager = (CacheManager)cm; + this.cacheStore = cachemanager.getCache(name, true); } /** @@ -68,6 +57,11 @@ */ public Cache get(Type type, CacheConfiguration config) { if (!destroyed) { + + if (!this.cacheStore.getCacheStatus().allowInvocations()) { + this.cacheStore.start(); + } + Node cacheRoot = this.cacheStore.getRoot().addChild(Fqn.fromString("Teiid")); //$NON-NLS-1$ Node node = cacheRoot.addChild(Fqn.fromString(type.location())); Modified: trunk/engine/src/main/java/org/teiid/cache/Cache.java =================================================================== --- trunk/engine/src/main/java/org/teiid/cache/Cache.java 2010-07-28 17:06:09 UTC (rev 2385) +++ trunk/engine/src/main/java/org/teiid/cache/Cache.java 2010-07-28 18:04:14 UTC (rev 2386) @@ -32,15 +32,9 @@ */ public interface Cache<K, V> { - public enum Type { REGISTRY("Registry"), //$NON-NLS-1$ - SESSION("Session"), //$NON-NLS-1$ - SESSION_MONITOR("Session-Monitor"), //$NON-NLS-1$ - AUTHORIZATION_POLICY("Authorization-Policy"), //$NON-NLS-1$ - AUTHORIZATION_PRINCIPAL("Auhtorization-Principal"), //$NON-NLS-1$ + public enum Type {SESSION("Session"), //$NON-NLS-1$ RESULTSET("ResultSet"), //$NON-NLS-1$ - VDBMETADATA("VdbMetadata"), //$NON-NLS-1$ - VDBMODELS("VdbModels"), //$NON-NLS-1$ - SCOPED_CACHE("Scoped-Cache"); //$NON-NLS-1$ + PREPAREDPLAN("PreparaedPlan"); //$NON-NLS-1$ private String location; Added: trunk/engine/src/main/java/org/teiid/cache/DefaultCache.java =================================================================== --- trunk/engine/src/main/java/org/teiid/cache/DefaultCache.java (rev 0) +++ trunk/engine/src/main/java/org/teiid/cache/DefaultCache.java 2010-07-28 18:04:14 UTC (rev 2386) @@ -0,0 +1,121 @@ +/* + * JBoss, Home of Professional Open Source. + * See the COPYRIGHT.txt file distributed with this work for information + * regarding copyright ownership. Some portions may be licensed + * to Red Hat, Inc. under one or more contributor license agreements. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA + * 02110-1301 USA. + */ +package org.teiid.cache; + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.teiid.core.util.LRUCache; + +public class DefaultCache<K, V> implements Cache<K, V>, Serializable { + private static final long serialVersionUID = -511120208522577206L; + public static final int DEFAULT_MAX_SIZE_TOTAL = 250; + + Map<K, V> map; + Map<String, Cache> children = new HashMap(); + String name; + + public DefaultCache(String name) { + this(name, DEFAULT_MAX_SIZE_TOTAL); + } + + public DefaultCache(String name, int maxSize) { + if(maxSize < 0){ + maxSize = DEFAULT_MAX_SIZE_TOTAL; + } + this.map = Collections.synchronizedMap(new LRUCache<K, V>(maxSize)); + this.name = name; + } + + public void addListener(CacheListener listener) { + } + + public void clear() { + map.clear(); + } + + public V get(K key) { + return map.get(key); + } + + public Set<K> keySet() { + return map.keySet(); + } + + public V put(K key, V value) { + return map.put(key, value); + } + + public V remove(K key) { + return map.remove(key); + } + + public int size() { + return map.size(); + } + + public Collection<V> values() { + return map.values(); + } + + @Override + public void removeListener() { + } + + @Override + public Cache addChild(String name) { + if (children.get(name) != null) { + return children.get(name); + } + + Cache c = new DefaultCache(name); + children.put(name, c); + return c; + } + + @Override + public Cache getChild(String name) { + return children.get(name); + } + + @Override + public List<Cache> getChildren() { + return new ArrayList<Cache>(children.values()); + } + + @Override + public boolean removeChild(String name) { + Object obj = children.remove(name); + return obj != null; + } + + @Override + public String getName() { + return name; + } +} \ No newline at end of file Property changes on: trunk/engine/src/main/java/org/teiid/cache/DefaultCache.java ___________________________________________________________________ Name: svn:mime-type + text/plain Modified: trunk/engine/src/main/java/org/teiid/cache/DefaultCacheFactory.java =================================================================== --- trunk/engine/src/main/java/org/teiid/cache/DefaultCacheFactory.java 2010-07-28 17:06:09 UTC (rev 2385) +++ trunk/engine/src/main/java/org/teiid/cache/DefaultCacheFactory.java 2010-07-28 18:04:14 UTC (rev 2386) @@ -22,12 +22,6 @@ package org.teiid.cache; import java.io.Serializable; -import java.util.ArrayList; -import java.util.Collection; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Set; import org.teiid.cache.Cache.Type; import org.teiid.core.TeiidRuntimeException; @@ -36,9 +30,13 @@ public class DefaultCacheFactory implements CacheFactory, Serializable { private static final long serialVersionUID = -5541424157695857527L; - DefaultCache cacheRoot = new DefaultCache("Teiid"); + DefaultCache cacheRoot; private volatile boolean destroyed = false; + public DefaultCacheFactory() { + this.cacheRoot = new DefaultCache("Teiid"); //$NON-NLS-1$ + } + @Override public void destroy() { this.destroyed = true; @@ -52,81 +50,4 @@ } throw new TeiidRuntimeException("Cache system has been shutdown"); //$NON-NLS-1$ } - - class DefaultCache<K, V> implements Cache<K, V>, Serializable { - Map<K, V> map = new HashMap(); - Map<String, Cache> children = new HashMap(); - String name; - - public DefaultCache(String name) { - this.name = name; - } - public void addListener(CacheListener listener) { - } - - public void clear() { - map.clear(); - } - - public V get(K key) { - return map.get(key); - } - - public Set<K> keySet() { - return map.keySet(); - } - - public V put(K key, V value) { - return map.put(key, value); - } - - public V remove(K key) { - return map.remove(key); - } - - public int size() { - return map.size(); - } - - public Collection<V> values() { - return map.values(); - } - - @Override - public void removeListener() { - } - - @Override - public Cache addChild(String name) { - if (children.get(name) != null) { - return children.get(name); - } - - Cache c = new DefaultCache(name); - children.put(name, c); - return c; - } - - @Override - public Cache getChild(String name) { - return children.get(name); - } - - @Override - public List<Cache> getChildren() { - return new ArrayList<Cache>(children.values()); - } - - @Override - public boolean removeChild(String name) { - Object obj = children.remove(name); - return obj != null; - } - - @Override - public String getName() { - return name; - } - } - } 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 2010-07-28 17:06:09 UTC (rev 2385) +++ trunk/engine/src/main/java/org/teiid/dqp/internal/process/CachedResults.java 2010-07-28 18:04:14 UTC (rev 2386) @@ -22,12 +22,15 @@ package org.teiid.dqp.internal.process; +import java.io.Serializable; + import org.teiid.common.buffer.TupleBuffer; import org.teiid.query.analysis.AnalysisRecord; import org.teiid.query.sql.lang.Command; -public class CachedResults { +public class CachedResults implements Serializable { + private static final long serialVersionUID = -5603182134635082207L; private Command command; private AnalysisRecord analysisRecord; private TupleBuffer results; Modified: trunk/engine/src/main/java/org/teiid/dqp/internal/process/DQPCore.java =================================================================== --- trunk/engine/src/main/java/org/teiid/dqp/internal/process/DQPCore.java 2010-07-28 17:06:09 UTC (rev 2385) +++ trunk/engine/src/main/java/org/teiid/dqp/internal/process/DQPCore.java 2010-07-28 18:04:14 UTC (rev 2386) @@ -44,6 +44,8 @@ import org.teiid.adminapi.Request.ThreadState; import org.teiid.adminapi.impl.RequestMetadata; import org.teiid.adminapi.impl.WorkerPoolStatisticsMetadata; +import org.teiid.cache.Cache; +import org.teiid.cache.CacheFactory; import org.teiid.client.DQP; import org.teiid.client.RequestMessage; import org.teiid.client.ResultsMessage; @@ -207,6 +209,7 @@ private int maxActivePlans = DQPConfiguration.DEFAULT_MAX_ACTIVE_PLANS; private int currentlyActivePlans; private LinkedList<RequestWorkItem> waitingPlans = new LinkedList<RequestWorkItem>(); + private CacheFactory cacheFactory; /** * perform a full shutdown and wait for 10 seconds for all threads to finish @@ -653,11 +656,11 @@ //result set cache if (config.isResultSetCacheEnabled()) { - this.rsCache = new SessionAwareCache<CachedResults>(config.getResultSetCacheMaxEntries()); + this.rsCache = new SessionAwareCache<CachedResults>(config.getResultSetCacheMaxEntries(), this.cacheFactory, Cache.Type.RESULTSET); } //prepared plan cache - prepPlanCache = new SessionAwareCache<PreparedPlan>(config.getPreparedPlanCacheMaxCount()); + prepPlanCache = new SessionAwareCache<PreparedPlan>(config.getPreparedPlanCacheMaxCount(), this.cacheFactory, Cache.Type.PREPAREDPLAN); // Processor debug flag LogManager.logInfo(LogConstants.CTX_DQP, DQPPlugin.Util.getString("DQPCore.Processor_debug_allowed_{0}", this.processorDebugAllowed)); //$NON-NLS-1$ @@ -831,4 +834,8 @@ return maxSourceRows; } + public void setCacheFactory(CacheFactory factory) { + this.cacheFactory = factory; + } + } \ No newline at end of file Modified: trunk/engine/src/main/java/org/teiid/dqp/internal/process/PreparedStatementRequest.java =================================================================== --- trunk/engine/src/main/java/org/teiid/dqp/internal/process/PreparedStatementRequest.java 2010-07-28 17:06:09 UTC (rev 2385) +++ trunk/engine/src/main/java/org/teiid/dqp/internal/process/PreparedStatementRequest.java 2010-07-28 18:04:14 UTC (rev 2386) @@ -146,7 +146,7 @@ // Defect 13751: Clone the plan in its current state (i.e. before processing) so that it can be used for later queries prepPlan.setPlan(processPlan.clone()); prepPlan.setAnalysisRecord(analysisRecord); - this.prepPlanCache.put(id, this.context.isSessionFunctionEvaluated(), prepPlan); + this.prepPlanCache.put(id, this.context.isSessionFunctionEvaluated(), this.context.isUserFunctionEvaluated(), prepPlan); } } else { LogManager.logTrace(LogConstants.CTX_DQP, new Object[] { "Query exist in cache: ", sqlQuery }); //$NON-NLS-1$ 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-07-28 17:06:09 UTC (rev 2385) +++ trunk/engine/src/main/java/org/teiid/dqp/internal/process/RequestWorkItem.java 2010-07-28 18:04:14 UTC (rev 2386) @@ -363,11 +363,12 @@ } if (doneProducingBatches && cid != null) { boolean sessionScope = processor.getContext().isSessionFunctionEvaluated(); + boolean userScope = processor.getContext().isUserFunctionEvaluated(); CachedResults cr = new CachedResults(); cr.setCommand(originalCommand); cr.setAnalysisRecord(analysisRecord); cr.setResults(resultsBuffer); - dqpCore.getRsCache().put(cid, sessionScope, cr); + dqpCore.getRsCache().put(cid, sessionScope, userScope, cr); } add = sendResultsIfNeeded(batch); if (!added) { 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 2010-07-28 17:06:09 UTC (rev 2385) +++ trunk/engine/src/main/java/org/teiid/dqp/internal/process/SessionAwareCache.java 2010-07-28 18:04:14 UTC (rev 2386) @@ -22,14 +22,18 @@ package org.teiid.dqp.internal.process; -import java.util.Collections; +import java.io.Serializable; +import java.util.ArrayList; import java.util.List; -import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; +import org.teiid.cache.Cache; +import org.teiid.cache.CacheConfiguration; +import org.teiid.cache.CacheFactory; +import org.teiid.cache.DefaultCache; +import org.teiid.cache.DefaultCacheFactory; import org.teiid.core.util.EquivalenceUtil; import org.teiid.core.util.HashCodeUtil; -import org.teiid.core.util.LRUCache; import org.teiid.query.parser.ParseInfo; import org.teiid.vdb.runtime.VDBKey; @@ -40,29 +44,48 @@ public class SessionAwareCache<T> { public static final int DEFAULT_MAX_SIZE_TOTAL = 250; - private Map<CacheID, T> cache; + private Cache<CacheID, T> localCache; + private Cache<CacheID, T> distributedCache; + private int maxSize; private AtomicInteger cacheHit = new AtomicInteger(); SessionAwareCache(){ - this(DEFAULT_MAX_SIZE_TOTAL); + this(DEFAULT_MAX_SIZE_TOTAL, new DefaultCacheFactory(), Cache.Type.RESULTSET); } - SessionAwareCache (int maxSize ){ + SessionAwareCache(int maxSize){ + this(maxSize, new DefaultCacheFactory(), Cache.Type.RESULTSET); + } + + SessionAwareCache (int maxSize, final CacheFactory cacheFactory, final Cache.Type type){ if(maxSize < 0){ - maxSize = DEFAULT_MAX_SIZE_TOTAL; + this.maxSize = DEFAULT_MAX_SIZE_TOTAL; } this.maxSize = maxSize; - cache = Collections.synchronizedMap(new LRUCache<CacheID, T>(maxSize)); + this.localCache = new DefaultCache<CacheID, T>("local", maxSize); //$NON-NLS-1$ + + this.distributedCache = localCache; + +// if (type == Cache.Type.PREPAREDPLAN) { +// this.distributedCache = localCache; +// } +// else { +// this.distributedCache = cacheFactory.get(type, new CacheConfiguration(CacheConfiguration.Policy.LRU, -1, SessionAwareCache.this.maxSize)); +// } } public T get(CacheID id){ id.setSessionId(id.originalSessionId); - T result = cache.get(id); + T result = localCache.get(id); if (result == null) { id.setSessionId(null); - result = cache.get(id); + result = distributedCache.get(id); + if (result == null) { + id.setUserName(null); + result = distributedCache.get(id); + } } if (result != null) { cacheHit.getAndIncrement(); @@ -77,13 +100,20 @@ /** * Create PreparedPlan for the given clientConn and SQl query */ - public void put(CacheID id, boolean sessionSpecific, T t){ + public void put(CacheID id, boolean sessionSpecific, boolean userSpecific, T t){ + if (!id.cachable) { + return; + } if (sessionSpecific) { id.setSessionId(id.originalSessionId); + this.localCache.put(id, t); } else { id.setSessionId(null); + if (!userSpecific) { + id.setUserName(null); + } + this.distributedCache.put(id, t); } - this.cache.put(id, t); } /** @@ -91,22 +121,27 @@ * @param clientConn ClientConnection */ public void clearAll(){ - cache.clear(); + this.localCache.clear(); + this.distributedCache.clear(); } - static class CacheID{ + static class CacheID implements Serializable { + private static final long serialVersionUID = 8261905111156764744L; private String sql; private VDBKey vdbInfo; private ParseInfo pi; private String sessionId; private String originalSessionId; - private List<?> parameters; + private List<Serializable> parameters; + private String userName; + boolean cachable = true; CacheID(DQPWorkContext context, ParseInfo pi, String sql){ this.sql = sql; this.vdbInfo = new VDBKey(context.getVdbName(), context.getVdbVersion()); this.pi = pi; this.originalSessionId = context.getSessionId(); + this.userName = context.getUserName(); } private void setSessionId(String sessionId) { @@ -114,8 +149,22 @@ } public void setParameters(List<?> parameters) { - this.parameters = parameters; + if (parameters != null && !parameters.isEmpty()) { + this.parameters = new ArrayList<Serializable>(); + for (Object obj:parameters) { + if (obj instanceof Serializable) { + this.parameters.add((Serializable)obj); + } + else{ + this.cachable = false; + } + } + } } + + public void setUserName(String name) { + this.userName = name; + } public boolean equals(Object obj){ if(obj == this) { @@ -126,6 +175,7 @@ } CacheID that = (CacheID)obj; return this.pi.equals(that.pi) && this.vdbInfo.equals(that.vdbInfo) && this.sql.equals(that.sql) + && EquivalenceUtil.areEqual(this.userName, that.userName) && EquivalenceUtil.areEqual(this.sessionId, that.sessionId) && EquivalenceUtil.areEqual(this.parameters, that.parameters); } @@ -143,7 +193,7 @@ //for testing purpose int getSpaceUsed() { - return cache.size(); + return localCache.size(); } int getSpaceAllowed() { return maxSize; Modified: trunk/engine/src/main/java/org/teiid/query/function/FunctionDescriptor.java =================================================================== --- trunk/engine/src/main/java/org/teiid/query/function/FunctionDescriptor.java 2010-07-28 17:06:09 UTC (rev 2385) +++ trunk/engine/src/main/java/org/teiid/query/function/FunctionDescriptor.java 2010-07-28 18:04:14 UTC (rev 2386) @@ -46,7 +46,8 @@ * descriptor. */ public class FunctionDescriptor implements Serializable, Cloneable { - + private static final long serialVersionUID = 5374103983118037242L; + private static final boolean ALLOW_NAN_INFINITY = PropertiesUtils.getBooleanProperty(System.getProperties(), "org.teiid.allowNanInfinity", false); //$NON-NLS-1$ private String name; @@ -238,6 +239,11 @@ throw new FunctionExecutionException(ErrorMessageKeys.FUNCTION_0002, QueryPlugin.Util.getString(ErrorMessageKeys.FUNCTION_0002, getName())); } + if (getDeterministic() == FunctionMethod.USER_DETERMINISTIC && values.length > 0 && values[0] instanceof CommandContext) { + CommandContext cc = (CommandContext)values[0]; + cc.setUserFunctionEvaluated(true); + } + if (getDeterministic() >= FunctionMethod.SESSION_DETERMINISTIC && values.length > 0 && values[0] instanceof CommandContext) { CommandContext cc = (CommandContext)values[0]; cc.setSessionFunctionEvaluated(true); Modified: trunk/engine/src/main/java/org/teiid/query/function/metadata/FunctionMethod.java =================================================================== --- trunk/engine/src/main/java/org/teiid/query/function/metadata/FunctionMethod.java 2010-07-28 17:06:09 UTC (rev 2385) +++ trunk/engine/src/main/java/org/teiid/query/function/metadata/FunctionMethod.java 2010-07-28 18:04:14 UTC (rev 2386) @@ -56,7 +56,8 @@ * @see FunctionCategoryConstants */ public class FunctionMethod implements Serializable { - + private static final long serialVersionUID = -8039086494296455152L; + private static final String NOT_ALLOWED = "NOT_ALLOWED"; //$NON-NLS-1$ private static final String ALLOWED = "ALLOWED"; //$NON-NLS-1$ private static final String REQUIRED = "REQUIRED"; //$NON-NLS-1$ @@ -83,9 +84,10 @@ */ public static final int DETERMINISTIC = 0; public static final int SERVER_DETERMINISTIC = 1; - public static final int SESSION_DETERMINISTIC = 2; - public static final int COMMAND_DETERMINISTIC = 3; - public static final int NONDETERMINISTIC = 4; + public static final int USER_DETERMINISTIC = 2; + public static final int SESSION_DETERMINISTIC = 3; + public static final int COMMAND_DETERMINISTIC = 4; + public static final int NONDETERMINISTIC = 5; private String name; Modified: trunk/engine/src/main/java/org/teiid/query/function/source/SystemSource.java =================================================================== --- trunk/engine/src/main/java/org/teiid/query/function/source/SystemSource.java 2010-07-28 17:06:09 UTC (rev 2385) +++ trunk/engine/src/main/java/org/teiid/query/function/source/SystemSource.java 2010-07-28 18:04:14 UTC (rev 2386) @@ -192,12 +192,12 @@ new FunctionParameter[] { new FunctionParameter("roleType", DataTypeManager.DefaultDataTypes.STRING, QueryPlugin.Util.getString("SystemSource.hasRole_param1")), //$NON-NLS-1$ //$NON-NLS-2$ new FunctionParameter("roleName", DataTypeManager.DefaultDataTypes.STRING, QueryPlugin.Util.getString("SystemSource.hasRole_param2"))}, //$NON-NLS-1$ //$NON-NLS-2$ - new FunctionParameter("result", DataTypeManager.DefaultDataTypes.BOOLEAN, QueryPlugin.Util.getString("SystemSource.hasRole_result")), false, FunctionMethod.SESSION_DETERMINISTIC ) ); //$NON-NLS-1$ //$NON-NLS-2$ + new FunctionParameter("result", DataTypeManager.DefaultDataTypes.BOOLEAN, QueryPlugin.Util.getString("SystemSource.hasRole_result")), false, FunctionMethod.USER_DETERMINISTIC ) ); //$NON-NLS-1$ //$NON-NLS-2$ functions.add(new FunctionMethod("hasRole", QueryPlugin.Util.getString("SystemSource.hasRole_description"), SECURITY, FunctionMethod.CANNOT_PUSHDOWN, SECURITY_FUNCTION_CLASS, "hasRole", //$NON-NLS-1$ //$NON-NLS-2$ //$NON-NLS-3$ new FunctionParameter[] { new FunctionParameter("roleName", DataTypeManager.DefaultDataTypes.STRING, QueryPlugin.Util.getString("SystemSource.hasRole_param2"))}, //$NON-NLS-1$ //$NON-NLS-2$ - new FunctionParameter("result", DataTypeManager.DefaultDataTypes.BOOLEAN, QueryPlugin.Util.getString("SystemSource.hasRole_result")), false, FunctionMethod.SESSION_DETERMINISTIC ) ); //$NON-NLS-1$ //$NON-NLS-2$ + new FunctionParameter("result", DataTypeManager.DefaultDataTypes.BOOLEAN, QueryPlugin.Util.getString("SystemSource.hasRole_result")), false, FunctionMethod.USER_DETERMINISTIC ) ); //$NON-NLS-1$ //$NON-NLS-2$ } private void addFormatNumberFunctions() { @@ -769,13 +769,15 @@ private void addUserFunction() { functions.add( new FunctionMethod("user", QueryPlugin.Util.getString("SystemSource.User_desc"), MISCELLANEOUS, FunctionMethod.CANNOT_PUSHDOWN, FUNCTION_CLASS, "user", null, //$NON-NLS-1$ //$NON-NLS-2$ //$NON-NLS-3$ - new FunctionParameter("result", DataTypeManager.DefaultDataTypes.STRING, QueryPlugin.Util.getString("SystemSource.User_result")), false, FunctionMethod.SESSION_DETERMINISTIC) ); //$NON-NLS-1$ //$NON-NLS-2$ + new FunctionParameter("result", DataTypeManager.DefaultDataTypes.STRING, QueryPlugin.Util.getString("SystemSource.User_result")), false, FunctionMethod.USER_DETERMINISTIC) ); //$NON-NLS-1$ //$NON-NLS-2$ } private void addCurrentDatabaseFunction() { + // this function is actually session_deterministic, however for caching purposes it has been set deterministic + // as caching is already scoped to the VDB. functions.add( new FunctionMethod("current_database", QueryPlugin.Util.getString("SystemSource.current_database_desc"), MISCELLANEOUS, FunctionMethod.CANNOT_PUSHDOWN, FUNCTION_CLASS, "current_database", null, //$NON-NLS-1$ //$NON-NLS-2$ //$NON-NLS-3$ - new FunctionParameter("result", DataTypeManager.DefaultDataTypes.STRING, QueryPlugin.Util.getString("current_database_result")), false, FunctionMethod.SESSION_DETERMINISTIC) ); //$NON-NLS-1$ //$NON-NLS-2$ + new FunctionParameter("result", DataTypeManager.DefaultDataTypes.STRING, QueryPlugin.Util.getString("current_database_result")), false, FunctionMethod.DETERMINISTIC) ); //$NON-NLS-1$ //$NON-NLS-2$ } private void addEnvFunction() { Modified: trunk/engine/src/main/java/org/teiid/query/parser/ParseInfo.java =================================================================== --- trunk/engine/src/main/java/org/teiid/query/parser/ParseInfo.java 2010-07-28 17:06:09 UTC (rev 2385) +++ trunk/engine/src/main/java/org/teiid/query/parser/ParseInfo.java 2010-07-28 18:04:14 UTC (rev 2386) @@ -22,12 +22,15 @@ package org.teiid.query.parser; +import java.io.Serializable; import java.util.HashMap; import java.util.Map; -public class ParseInfo { +public class ParseInfo implements Serializable{ + private static final long serialVersionUID = -7323683731955992888L; + public Map<String, Integer> nameCounts = new HashMap<String, Integer>(); public int referenceCount = 0; Modified: trunk/engine/src/main/java/org/teiid/query/util/CommandContext.java =================================================================== --- trunk/engine/src/main/java/org/teiid/query/util/CommandContext.java 2010-07-28 17:06:09 UTC (rev 2385) +++ trunk/engine/src/main/java/org/teiid/query/util/CommandContext.java 2010-07-28 18:04:14 UTC (rev 2386) @@ -90,6 +90,8 @@ private boolean sessionFunctionEvaluated; + private boolean userFunctionEvaluated; + private Set<String> groups; private long timeSliceEnd = Long.MAX_VALUE; @@ -147,6 +149,14 @@ globalState.sessionFunctionEvaluated = sessionFunctionEvaluated; } + public boolean isUserFunctionEvaluated() { + return globalState.userFunctionEvaluated; + } + + public void setUserFunctionEvaluated(boolean userFunctionEvaluated) { + globalState.userFunctionEvaluated = userFunctionEvaluated; + } + /** * @return */ 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 2010-07-28 17:06:09 UTC (rev 2385) +++ trunk/engine/src/test/java/org/teiid/dqp/internal/process/TestPreparedPlanCache.java 2010-07-28 18:04:14 UTC (rev 2386) @@ -61,7 +61,7 @@ //No PreparedPlan at the begining assertNull(cache.get(id)); //create one - cache.put(id, true, new PreparedPlan()); + cache.put(id, true, false, new PreparedPlan()); //should have one now assertNotNull("Unable to get prepared plan from cache", cache.get(id)); //$NON-NLS-1$ } @@ -148,7 +148,7 @@ CacheID id = new CacheID(session, pi, dummy.toString()); PreparedPlan pPlan = new PreparedPlan(); - cache.put(id, true, pPlan); + cache.put(id, true, false, pPlan); pPlan.setCommand(dummy); pPlan.setPlan(new RelationalPlan(new ProjectNode(i))); AnalysisRecord analysisRecord = new AnalysisRecord(true, false); Modified: trunk/jboss-integration/pom.xml =================================================================== --- trunk/jboss-integration/pom.xml 2010-07-28 17:06:09 UTC (rev 2385) +++ trunk/jboss-integration/pom.xml 2010-07-28 18:04:14 UTC (rev 2386) @@ -22,9 +22,14 @@ <groupId>org.jboss.teiid</groupId> <artifactId>teiid-api</artifactId> </dependency> - + <dependency> <groupId>org.jboss.teiid</groupId> + <artifactId>teiid-cache-jbosscache</artifactId> + </dependency> + + <dependency> + <groupId>org.jboss.teiid</groupId> <artifactId>teiid-client</artifactId> <scope>provided</scope> </dependency> Modified: trunk/jboss-integration/src/main/java/org/teiid/jboss/deployers/RuntimeEngineDeployer.java =================================================================== --- trunk/jboss-integration/src/main/java/org/teiid/jboss/deployers/RuntimeEngineDeployer.java 2010-07-28 17:06:09 UTC (rev 2385) +++ trunk/jboss-integration/src/main/java/org/teiid/jboss/deployers/RuntimeEngineDeployer.java 2010-07-28 18:04:14 UTC (rev 2386) @@ -54,6 +54,7 @@ import org.teiid.adminapi.impl.SessionMetadata; import org.teiid.adminapi.impl.WorkerPoolStatisticsMetadata; import org.teiid.adminapi.jboss.AdminProvider; +import org.teiid.cache.CacheFactory; import org.teiid.client.DQP; import org.teiid.client.security.ILogon; import org.teiid.client.util.ExceptionUtil; @@ -128,28 +129,43 @@ // create the necessary services createClientServices(); + int offset = 0; + String portBinding = System.getProperty("jboss.service.binding.set"); //$NON-NLS-1$ + if (portBinding != null && portBinding.startsWith("ports-")) { //$NON-NLS-1$ + if (portBinding.equals("ports-default")) { //$NON-NLS-1$ + offset = 0; + } + else { + try { + offset = Integer.parseInt(portBinding.substring(portBinding.length()-2))*100; + } catch (NumberFormatException e) { + offset = 0; + } + } + } + this.csr.registerClientService(ILogon.class, logon, LogConstants.CTX_SECURITY); this.csr.registerClientService(DQP.class, proxyService(DQP.class, this.dqpCore, LogConstants.CTX_DQP), LogConstants.CTX_DQP); this.csr.registerClientService(Admin.class, proxyService(Admin.class, admin, LogConstants.CTX_ADMIN_API), LogConstants.CTX_ADMIN_API); if (this.jdbcSocketConfiguration.isEnabled()) { - this.jdbcSocket = new SocketListener(this.jdbcSocketConfiguration, csr, this.dqpCore.getBufferManager()); - LogManager.logInfo(LogConstants.CTX_RUNTIME, IntegrationPlugin.Util.getString("socket_enabled","Teiid JDBC = ",(this.jdbcSocketConfiguration.getSSLConfiguration().isSslEnabled()?"mms://":"mm://")+this.jdbcSocketConfiguration.getHostAddress().getHostName()+":"+this.jdbcSocketConfiguration.getPortNumber())); //$NON-NLS-1$ //$NON-NLS-2$ //$NON-NLS-3$ //$NON-NLS-4$ //$NON-NLS-5$ + this.jdbcSocket = new SocketListener(this.jdbcSocketConfiguration, csr, this.dqpCore.getBufferManager(), offset); + LogManager.logInfo(LogConstants.CTX_RUNTIME, IntegrationPlugin.Util.getString("socket_enabled","Teiid JDBC = ",(this.jdbcSocketConfiguration.getSSLConfiguration().isSslEnabled()?"mms://":"mm://")+this.jdbcSocketConfiguration.getHostAddress().getHostName()+":"+this.jdbcSocketConfiguration.getPortNumber()+offset)); //$NON-NLS-1$ //$NON-NLS-2$ //$NON-NLS-3$ //$NON-NLS-4$ //$NON-NLS-5$ } else { LogManager.logInfo(LogConstants.CTX_RUNTIME, IntegrationPlugin.Util.getString("socket_not_enabled", "jdbc connections")); //$NON-NLS-1$ //$NON-NLS-2$ } if (this.adminSocketConfiguration.isEnabled()) { - this.adminSocket = new SocketListener(this.adminSocketConfiguration, csr, this.dqpCore.getBufferManager()); - LogManager.logInfo(LogConstants.CTX_RUNTIME, IntegrationPlugin.Util.getString("socket_enabled","Teiid Admin", (this.adminSocketConfiguration.getSSLConfiguration().isSslEnabled()?"mms://":"mm://")+this.adminSocketConfiguration.getHostAddress().getHostName()+":"+this.adminSocketConfiguration.getPortNumber())); //$NON-NLS-1$ //$NON-NLS-2$ //$NON-NLS-3$ //$NON-NLS-4$ //$NON-NLS-5$ + this.adminSocket = new SocketListener(this.adminSocketConfiguration, csr, this.dqpCore.getBufferManager(), offset); + LogManager.logInfo(LogConstants.CTX_RUNTIME, IntegrationPlugin.Util.getString("socket_enabled","Teiid Admin", (this.adminSocketConfiguration.getSSLConfiguration().isSslEnabled()?"mms://":"mm://")+this.adminSocketConfiguration.getHostAddress().getHostName()+":"+this.adminSocketConfiguration.getPortNumber()+offset)); //$NON-NLS-1$ //$NON-NLS-2$ //$NON-NLS-3$ //$NON-NLS-4$ //$NON-NLS-5$ } else { LogManager.logInfo(LogConstants.CTX_RUNTIME, IntegrationPlugin.Util.getString("socket_not_enabled", "admin connections")); //$NON-NLS-1$ //$NON-NLS-2$ } if (this.odbcSocketConfiguration.isEnabled()) { this.vdbRepository.odbcEnabled(); - this.odbcSocket = new ODBCSocketListener(this.odbcSocketConfiguration, csr, this.dqpCore.getBufferManager()); - LogManager.logInfo(LogConstants.CTX_RUNTIME, IntegrationPlugin.Util.getString("odbc_enabled","Teiid ODBC - SSL=", (this.odbcSocketConfiguration.getSSLConfiguration().isSslEnabled()?"ON":"OFF")+" Host = "+this.odbcSocketConfiguration.getHostAddress().getHostName()+" Port = "+this.odbcSocketConfiguration.getPortNumber())); //$NON-NLS-1$ //$NON-NLS-2$ //$NON-NLS-3$ //$NON-NLS-4$ //$NON-NLS-5$ //$NON-NLS-6$ + this.odbcSocket = new ODBCSocketListener(this.odbcSocketConfiguration, csr, this.dqpCore.getBufferManager(), offset); + LogManager.logInfo(LogConstants.CTX_RUNTIME, IntegrationPlugin.Util.getString("odbc_enabled","Teiid ODBC - SSL=", (this.odbcSocketConfiguration.getSSLConfiguration().isSslEnabled()?"ON":"OFF")+" Host = "+this.odbcSocketConfiguration.getHostAddress().getHostName()+" Port = "+this.odbcSocketConfiguration.getPortNumber()+offset)); //$NON-NLS-1$ //$NON-NLS-2$ //$NON-NLS-3$ //$NON-NLS-4$ //$NON-NLS-5$ //$NON-NLS-6$ } else { LogManager.logInfo(LogConstants.CTX_RUNTIME, IntegrationPlugin.Util.getString("odbc_not_enabled")); //$NON-NLS-1$ } @@ -408,4 +424,8 @@ String targetVDBName, int targetVDBVersion) throws AdminException { this.vdbRepository.mergeVDBs(sourceVDBName, sourceVDBVersion, targetVDBName, targetVDBVersion); } + + public void setCacheFactory(CacheFactory factory) { + this.dqpCore.setCacheFactory(factory); + } } Modified: trunk/runtime/src/main/java/org/teiid/transport/ODBCSocketListener.java =================================================================== --- trunk/runtime/src/main/java/org/teiid/transport/ODBCSocketListener.java 2010-07-28 17:06:09 UTC (rev 2385) +++ trunk/runtime/src/main/java/org/teiid/transport/ODBCSocketListener.java 2010-07-28 18:04:14 UTC (rev 2386) @@ -33,8 +33,8 @@ public class ODBCSocketListener extends SocketListener { private ODBCServerRemote.AuthenticationType authType = ODBCServerRemote.AuthenticationType.CLEARTEXT; - public ODBCSocketListener(SocketConfiguration config, ClientServiceRegistryImpl csr, StorageManager storageManager) { - super(config, csr, storageManager); + public ODBCSocketListener(SocketConfiguration config, ClientServiceRegistryImpl csr, StorageManager storageManager, int portOffset) { + super(config, csr, storageManager, portOffset); } @Override Modified: trunk/runtime/src/main/java/org/teiid/transport/SocketListener.java =================================================================== --- trunk/runtime/src/main/java/org/teiid/transport/SocketListener.java 2010-07-28 17:06:09 UTC (rev 2385) +++ trunk/runtime/src/main/java/org/teiid/transport/SocketListener.java 2010-07-28 18:04:14 UTC (rev 2386) @@ -51,8 +51,8 @@ private ExecutorService nettyPool; private ClientServiceRegistryImpl csr; - public SocketListener(SocketConfiguration config, ClientServiceRegistryImpl csr, StorageManager storageManager) { - this(config.getPortNumber(), config.getHostAddress().getHostAddress(), config.getInputBufferSize(), config.getOutputBufferSize(), config.getMaxSocketThreads(), config.getSSLConfiguration(), csr, storageManager); + public SocketListener(SocketConfiguration config, ClientServiceRegistryImpl csr, StorageManager storageManager, int portOffset) { + this(config.getPortNumber()+portOffset, config.getHostAddress().getHostAddress(), config.getInputBufferSize(), config.getOutputBufferSize(), config.getMaxSocketThreads(), config.getSSLConfiguration(), csr, storageManager); LogManager.logDetail(LogConstants.CTX_TRANSPORT, RuntimePlugin.Util.getString("SocketTransport.1", new Object[] {config.getHostAddress().getHostAddress(), String.valueOf(config.getPortNumber())})); //$NON-NLS-1$ }