[teiid-commits] teiid SVN: r2595 - in branches/7.1.x/engine/src/main/java/org/teiid: query/tempdata and 1 other directory.

[teiid-commits at lists.jboss.org]

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
 	 */