Author: shawkins Date: 2010-09-22 12:59:20 -0400 (Wed, 22 Sep 2010) New Revision: 2595 Modified: branches/7.1.x/engine/src/main/java/org/teiid/dqp/internal/process/DQPCore.java branches/7.1.x/engine/src/main/java/org/teiid/query/tempdata/TempTableDataManager.java Log: TEIID-1212 adding replication support for internal materialization Modified: branches/7.1.x/engine/src/main/java/org/teiid/dqp/internal/process/DQPCore.java =================================================================== --- branches/7.1.x/engine/src/main/java/org/teiid/dqp/internal/process/DQPCore.java 2010-09-22 16:46:21 UTC (rev 2594) +++ branches/7.1.x/engine/src/main/java/org/teiid/dqp/internal/process/DQPCore.java 2010-09-22 16:59:20 UTC (rev 2595) @@ -211,6 +211,8 @@ private int currentlyActivePlans; private LinkedList<RequestWorkItem> waitingPlans = new LinkedList<RequestWorkItem>(); private CacheFactory cacheFactory; + + private SessionAwareCache<CachedResults> matTables; /** * perform a full shutdown and wait for 10 seconds for all threads to finish @@ -615,6 +617,9 @@ clearResultSetCache(vdbName, version); break; } + if (this.matTables != null) { + this.matTables.clearForVDB(vdbName, version); + } } public Collection<org.teiid.adminapi.Transaction> getTransactions() { @@ -700,7 +705,6 @@ this.processWorkerPool = new ThreadReuseExecutor(DQPConfiguration.PROCESS_PLAN_QUEUE_NAME, config.getMaxThreads()); - SessionAwareCache<CachedResults> matTables = null; if (cacheFactory.isReplicated()) { matTables = new SessionAwareCache<CachedResults>(); } Modified: branches/7.1.x/engine/src/main/java/org/teiid/query/tempdata/TempTableDataManager.java =================================================================== --- branches/7.1.x/engine/src/main/java/org/teiid/query/tempdata/TempTableDataManager.java 2010-09-22 16:46:21 UTC (rev 2594) +++ branches/7.1.x/engine/src/main/java/org/teiid/query/tempdata/TempTableDataManager.java 2010-09-22 16:59:20 UTC (rev 2595) @@ -138,7 +138,8 @@ private static class MatTableEntry implements Serializable { private static final long serialVersionUID = 8559613701442751579L; - transient long lastUpdate = System.currentTimeMillis(); + long lastUpdate = System.currentTimeMillis(); + boolean valid; } private Cache<MatTableKey, MatTableEntry> tables; @@ -156,7 +157,7 @@ tables = cacheFactory.get(Cache.Type.MATTABLES, cc); } } - + public TupleSource registerRequest( CommandContext context, Command command, @@ -303,6 +304,9 @@ LogManager.logDetail(LogConstants.CTX_MATVIEWS, "processing refreshmatview for", matViewName); //$NON-NLS-1$ MatTableInfo info = globalStore.getMatTableInfo(matTableName); boolean invalidate = Boolean.TRUE.equals(((Constant)proc.getParameter(1).getExpression()).getValue()); + if (invalidate) { + touchTable(context, matTableName, false); + } MatState oldState = info.setState(MatState.NEEDS_LOADING, invalidate?Boolean.FALSE:null, null); if (oldState == MatState.LOADING) { return CollectionTupleSource.createUpdateCountTupleSource(-1); @@ -395,7 +399,7 @@ if (entry != null && entry.lastUpdate > info.getUpdateTime() && info.getState() != MatState.LOADING) { //remote load - info.setState(MatState.NEEDS_LOADING, null, null); + info.setState(MatState.NEEDS_LOADING, entry.valid, null); loadTime = entry.lastUpdate; } } @@ -497,12 +501,7 @@ BatchCollector bc = qp.createBatchCollector(); TupleBuffer tb = bc.collectTuples(); cr.setResults(tb); - MatTableKey key = new MatTableKey(); - key.name = fullName; - key.vdb = new VDBKey(context.getVdbName(), context.getVdbVersion()); - MatTableEntry matTableEntry = new MatTableEntry(); - matTableEntry.lastUpdate = 0; - tables.put(key, matTableEntry, null); + touchTable(context, fullName, true); this.distributedCache.put(cid, FunctionMethod.VDB_DETERMINISTIC, cr, info.getTtl()); ts = tb.createIndexedTupleSource(); } else { @@ -541,6 +540,15 @@ return rowCount; } + private void touchTable(CommandContext context, String fullName, boolean valid) { + MatTableKey key = new MatTableKey(); + key.name = fullName; + key.vdb = new VDBKey(context.getVdbName(), context.getVdbVersion()); + MatTableEntry matTableEntry = new MatTableEntry(); + matTableEntry.valid = valid; + tables.put(key, matTableEntry, null); + } + /** * Return a list of ElementSymbols for the given index/key object */