Author: rareddy Date: 2011-07-27 17:30:01 -0400 (Wed, 27 Jul 2011) New Revision: 3345 Modified: branches/7.4.x/api/src/main/java/org/teiid/events/EventDistributor.java branches/7.4.x/build/kits/jboss-container/deploy/teiid/teiid-cache-manager-jboss-beans-rename-me.xml branches/7.4.x/cache-jbosscache/src/main/java/org/teiid/cache/jboss/JBossCacheFactory.java branches/7.4.x/cache-jbosscache/src/main/java/org/teiid/cache/jboss/TupleBatchCacheLoader.java branches/7.4.x/engine/src/main/java/org/teiid/common/buffer/TupleBuffer.java branches/7.4.x/engine/src/main/java/org/teiid/dqp/internal/process/CachedResults.java branches/7.4.x/engine/src/main/java/org/teiid/dqp/internal/process/DQPCore.java branches/7.4.x/engine/src/main/java/org/teiid/query/tempdata/TempTableDataManager.java branches/7.4.x/engine/src/main/java/org/teiid/query/tempdata/TempTableStore.java branches/7.4.x/engine/src/main/resources/org/teiid/query/i18n.properties branches/7.4.x/jboss-integration/src/main/java/org/teiid/jboss/deployers/RuntimeEngineDeployer.java branches/7.4.x/jboss-integration/src/main/resources/org/teiid/jboss/i18n.properties Log: TEIID-1657: added distributed refresh of cache node. When a internal materialized view(imv) is loaded at a node, after that imv contents finished loading at that node, a event will be sent to all the other nodes to refresh their content with contents of the original node. The other nodes will pull the updated data. Modified: branches/7.4.x/api/src/main/java/org/teiid/events/EventDistributor.java =================================================================== --- branches/7.4.x/api/src/main/java/org/teiid/events/EventDistributor.java 2011-07-27 21:13:25 UTC (rev 3344) +++ branches/7.4.x/api/src/main/java/org/teiid/events/EventDistributor.java 2011-07-27 21:30:01 UTC (rev 3345) @@ -121,4 +121,11 @@ */ void setViewDefinition(String vdbName, int vdbVersion, String schema, String viewName, String definition); + /** + * + * @param vdbName + * @param vdbVersion + * @param viewName + */ + void refreshMatView(String vdbName, int vdbVersion, String tableName); } Modified: branches/7.4.x/build/kits/jboss-container/deploy/teiid/teiid-cache-manager-jboss-beans-rename-me.xml =================================================================== --- branches/7.4.x/build/kits/jboss-container/deploy/teiid/teiid-cache-manager-jboss-beans-rename-me.xml 2011-07-27 21:13:25 UTC (rev 3344) +++ branches/7.4.x/build/kits/jboss-container/deploy/teiid/teiid-cache-manager-jboss-beans-rename-me.xml 2011-07-27 21:30:01 UTC (rev 3345) @@ -145,7 +145,7 @@ </property> <property name="clusterName">${jboss.partition.name:DefaultPartition}-teiid-events</property> <property name="multiplexerStack">${jboss.default.jgroups.stack:udp}</property> - <property name="localEventDistributorName">teiid/runtime-engine</property> + <property name="localEventDistributorName">teiid/engine-deployer</property> </bean> </deployment> \ No newline at end of file Modified: branches/7.4.x/cache-jbosscache/src/main/java/org/teiid/cache/jboss/JBossCacheFactory.java =================================================================== --- branches/7.4.x/cache-jbosscache/src/main/java/org/teiid/cache/jboss/JBossCacheFactory.java 2011-07-27 21:13:25 UTC (rev 3344) +++ branches/7.4.x/cache-jbosscache/src/main/java/org/teiid/cache/jboss/JBossCacheFactory.java 2011-07-27 21:30:01 UTC (rev 3345) @@ -59,7 +59,9 @@ if (!this.cacheStore.getCacheStatus().allowInvocations()) { this.cacheStore.start(); - this.cacheStore.getRegion(this.cacheStore.getRoot().getFqn(), true).activate(); + if (this.cacheStore.getRegion(this.cacheStore.getRoot().getFqn(), false) != null) { + this.cacheStore.getRegion(this.cacheStore.getRoot().getFqn(), true).activate(); + } } Node cacheRoot = this.cacheStore.getRoot().addChild(Fqn.fromString("Teiid")); //$NON-NLS-1$ Modified: branches/7.4.x/cache-jbosscache/src/main/java/org/teiid/cache/jboss/TupleBatchCacheLoader.java =================================================================== --- branches/7.4.x/cache-jbosscache/src/main/java/org/teiid/cache/jboss/TupleBatchCacheLoader.java 2011-07-27 21:13:25 UTC (rev 3344) +++ branches/7.4.x/cache-jbosscache/src/main/java/org/teiid/cache/jboss/TupleBatchCacheLoader.java 2011-07-27 21:30:01 UTC (rev 3345) @@ -62,9 +62,8 @@ map.put(id, b); return map; } - return super.get(fqn); } - return null; + return super.get(fqn); } @Override Modified: branches/7.4.x/engine/src/main/java/org/teiid/common/buffer/TupleBuffer.java =================================================================== --- branches/7.4.x/engine/src/main/java/org/teiid/common/buffer/TupleBuffer.java 2011-07-27 21:13:25 UTC (rev 3344) +++ branches/7.4.x/engine/src/main/java/org/teiid/common/buffer/TupleBuffer.java 2011-07-27 21:30:01 UTC (rev 3345) @@ -99,8 +99,12 @@ this.uuid = java.util.UUID.randomUUID().toString(); } return this.uuid; - } + } + public void setId(String uuid) { + this.uuid = uuid; + } + public boolean isLobs() { return lobIndexes != null; } Modified: branches/7.4.x/engine/src/main/java/org/teiid/dqp/internal/process/CachedResults.java =================================================================== --- branches/7.4.x/engine/src/main/java/org/teiid/dqp/internal/process/CachedResults.java 2011-07-27 21:13:25 UTC (rev 3344) +++ branches/7.4.x/engine/src/main/java/org/teiid/dqp/internal/process/CachedResults.java 2011-07-27 21:30:01 UTC (rev 3345) @@ -139,6 +139,7 @@ } buffer = bufferManager.createTupleBuffer(schema, "cached", TupleSourceType.FINAL); //$NON-NLS-1$ buffer.setBatchSize(this.batchSize); + buffer.setId(this.uuid); if (this.hint != null) { buffer.setPrefersMemory(this.hint.getPrefersMemory()); } @@ -151,6 +152,7 @@ return false; } buffer.addTupleBatch(batch, true); + cache.remove(uuid+","+row); //$NON-NLS-1$ } this.results = buffer; bufferManager.addTupleBuffer(this.results); Modified: branches/7.4.x/engine/src/main/java/org/teiid/dqp/internal/process/DQPCore.java =================================================================== --- branches/7.4.x/engine/src/main/java/org/teiid/dqp/internal/process/DQPCore.java 2011-07-27 21:13:25 UTC (rev 3344) +++ branches/7.4.x/engine/src/main/java/org/teiid/dqp/internal/process/DQPCore.java 2011-07-27 21:30:01 UTC (rev 3345) @@ -405,7 +405,7 @@ } } - void addWork(Runnable work) { + public void addWork(Runnable work) { this.processWorkerPool.execute(work); } Modified: branches/7.4.x/engine/src/main/java/org/teiid/query/tempdata/TempTableDataManager.java =================================================================== --- branches/7.4.x/engine/src/main/java/org/teiid/query/tempdata/TempTableDataManager.java 2011-07-27 21:13:25 UTC (rev 3344) +++ branches/7.4.x/engine/src/main/java/org/teiid/query/tempdata/TempTableDataManager.java 2011-07-27 21:30:01 UTC (rev 3345) @@ -309,15 +309,27 @@ QueryMetadataInterface metadata = context.getMetadata(); TempTableStore globalStore = context.getGlobalTableStore(); if (StringUtil.endsWithIgnoreCase(proc.getProcedureCallableName(), REFRESHMATVIEW)) { - Object groupID = validateMatView(metadata, proc); + Object groupID = validateMatView(metadata, (String)((Constant)proc.getParameter(1).getExpression()).getValue()); Object matTableId = context.getGlobalTableStore().getGlobalTempTableMetadataId(groupID, metadata); String matViewName = metadata.getFullName(groupID); String matTableName = metadata.getFullName(matTableId); LogManager.logDetail(LogConstants.CTX_MATVIEWS, "processing refreshmatview for", matViewName); //$NON-NLS-1$ MatTableInfo info = globalStore.getMatTableInfo(matTableName); + Long loadTime = null; + boolean useCache = false; + if (this.distributedCache != null) { + MatTableKey key = new MatTableKey(); + key.name = matTableName; + key.vdb = new VDBKey(context.getVdbName(),context.getVdbVersion()); + MatTableEntry entry = this.tables.get(key); + useCache = (entry != null && entry.valid && entry.lastUpdate > info.getUpdateTime()); + if (useCache) { + loadTime = entry.lastUpdate; + } + } boolean invalidate = Boolean.TRUE.equals(((Constant)proc.getParameter(2).getExpression()).getValue()); if (invalidate) { - touchTable(context, matTableName, false); + touchTable(context, matTableName, false, System.currentTimeMillis()); } MatState oldState = info.setState(MatState.NEEDS_LOADING, invalidate?Boolean.FALSE:null, null); if (oldState == MatState.LOADING) { @@ -325,10 +337,10 @@ } GroupSymbol matTable = new GroupSymbol(matTableName); matTable.setMetadataID(matTableId); - int rowCount = loadGlobalTable(context, matTable, matTableName, globalStore, info, null, false); + int rowCount = loadGlobalTable(context, matTable, matTableName, matViewName, globalStore, info, invalidate?null:loadTime, !invalidate && useCache); return CollectionTupleSource.createUpdateCountTupleSource(rowCount); } else if (StringUtil.endsWithIgnoreCase(proc.getProcedureCallableName(), REFRESHMATVIEWROW)) { - Object groupID = validateMatView(metadata, proc); + Object groupID = validateMatView(metadata, (String)((Constant)proc.getParameter(1).getExpression()).getValue()); Object pk = metadata.getPrimaryKey(groupID); String matViewName = metadata.getFullName(groupID); if (pk == null) { @@ -377,15 +389,36 @@ TempTable tempTable = globalStore.getOrCreateTempTable(matTableName, new Query(), bufferManager, false); return tempTable.updateTuple(tuple, delete); } + + public void refreshMatView(String vdbName, int vdbVersion, String viewName, + QueryMetadataInterface metadata, TempTableStore globalStore) + throws QueryProcessingException, TeiidComponentException, TeiidProcessingException { + + Object groupID = validateMatView(metadata, viewName); + Object matTableId = globalStore.getGlobalTempTableMetadataId(groupID, metadata); + String matViewName = metadata.getFullName(groupID); + String matTableName = metadata.getFullName(matTableId); + LogManager.logDetail(LogConstants.CTX_MATVIEWS, "processing refreshmatview for", matViewName); //$NON-NLS-1$ + MatTableInfo info = globalStore.getMatTableInfo(matTableName); - private Object validateMatView(QueryMetadataInterface metadata, - StoredProcedure proc) throws TeiidComponentException, + MatState oldState = info.setState(MatState.NEEDS_LOADING, Boolean.FALSE, null); + if (oldState == MatState.LOADING) { + return; + } + GroupSymbol matTable = new GroupSymbol(matTableName); + matTable.setMetadataID(matTableId); + CommandContext context = new CommandContext(new Object(), "internal", "internal", vdbName, vdbVersion); //$NON-NLS-1$ //$NON-NLS-2$ + context.setMetadata(metadata); + context.setGlobalTableStore(globalStore); + loadGlobalTable(context, matTable, matTableName, matViewName, globalStore, info, null, true); + } + + private Object validateMatView(QueryMetadataInterface metadata, String viewName) throws TeiidComponentException, TeiidProcessingException { - String name = (String)((Constant)proc.getParameter(1).getExpression()).getValue(); try { - Object groupID = metadata.getGroupID(name); + Object groupID = metadata.getGroupID(viewName); if (!metadata.hasMaterialization(groupID) || metadata.getMaterialization(groupID) != null) { - throw new QueryProcessingException(QueryPlugin.Util.getString("TempTableDataManager.not_implicit_matview", name)); //$NON-NLS-1$ + throw new QueryProcessingException(QueryPlugin.Util.getString("TempTableDataManager.not_implicit_matview", viewName)); //$NON-NLS-1$ } return groupID; } catch (QueryMetadataException e) { @@ -402,8 +435,13 @@ if (!group.isTempGroupSymbol()) { return null; } + String viewName = null; final String tableName = group.getNonCorrelationName().toUpperCase(); boolean remapColumns = !tableName.equalsIgnoreCase(group.getName()); + TempMetadataID groupID = (TempMetadataID)group.getMetadataID(); + if (groupID.getOriginalMetadataID() != null) { + viewName = context.getMetadata().getFullName(groupID.getOriginalMetadataID()); + } TempTable table = null; if (group.isGlobalTable()) { final TempTableStore globalStore = context.getGlobalTableStore(); @@ -429,9 +467,9 @@ if (load) { if (!info.isValid()) { //blocking load - loadGlobalTable(context, group, tableName, globalStore, info, loadTime, true); + loadGlobalTable(context, group, tableName, viewName, globalStore, info, loadTime, true); } else { - loadAsynch(context, group, tableName, globalStore, info, loadTime); + loadAsynch(context, group, tableName, viewName, globalStore, info, loadTime); } } table = globalStore.getOrCreateTempTable(tableName, query, bufferManager, false); @@ -464,13 +502,13 @@ } private void loadAsynch(final CommandContext context, - final GroupSymbol group, final String tableName, + final GroupSymbol group, final String tableName, final String viewName, final TempTableStore globalStore, final MatTableInfo info, final Long loadTime) { Callable<Integer> toCall = new Callable<Integer>() { @Override public Integer call() throws Exception { - return loadGlobalTable(context, group, tableName, globalStore, info, loadTime, true); + return loadGlobalTable(context, group, tableName, viewName, globalStore, info, loadTime, true); } }; FutureTask<Integer> task = new FutureTask<Integer>(toCall); @@ -478,7 +516,7 @@ } private int loadGlobalTable(CommandContext context, - GroupSymbol group, final String tableName, + GroupSymbol group, final String tableName, final String viewName, TempTableStore globalStore, MatTableInfo info, Long loadTime, boolean useCache) throws TeiidComponentException, TeiidProcessingException { LogManager.logInfo(LogConstants.CTX_MATVIEWS, QueryPlugin.Util.getString("TempTableDataManager.loading", tableName)); //$NON-NLS-1$ @@ -506,6 +544,7 @@ } } int rowCount = -1; + boolean viewFetched = false; try { String fullName = metadata.getFullName(group.getMetadataID()); TupleSource ts = null; @@ -517,6 +556,7 @@ CachedResults cr = this.distributedCache.get(cid); if (cr != null) { ts = cr.getResults().createIndexedTupleSource(); + LogManager.logInfo(LogConstants.CTX_MATVIEWS, QueryPlugin.Util.getString("TempTableDataManager.cache_load", tableName)); //$NON-NLS-1$ } } } @@ -536,9 +576,10 @@ BatchCollector bc = qp.createBatchCollector(); TupleBuffer tb = bc.collectTuples(); cr.setResults(tb, qp.getProcessorPlan()); - touchTable(context, fullName, true); + touchTable(context, fullName, true, info.getUpdateTime()); this.distributedCache.put(cid, Determinism.VDB_DETERMINISTIC, cr, info.getTtl()); ts = tb.createIndexedTupleSource(); + viewFetched = true; } else { ts = new BatchCollector.BatchProducerTupleSource(qp); } @@ -570,18 +611,22 @@ } else { globalStore.swapTempTable(tableName, table); info.setState(MatState.LOADED, true, loadTime); + if (viewFetched & viewName != null && this.eventDistributor != null) { + this.eventDistributor.refreshMatView(context.getVdbName(), context.getVdbVersion(), viewName); + } LogManager.logInfo(LogConstants.CTX_MATVIEWS, QueryPlugin.Util.getString("TempTableDataManager.loaded", tableName, rowCount)); //$NON-NLS-1$ } } return rowCount; } - private void touchTable(CommandContext context, String fullName, boolean valid) { + private void touchTable(CommandContext context, String fullName, boolean valid, long loadtime) { MatTableKey key = new MatTableKey(); key.name = fullName; key.vdb = new VDBKey(context.getVdbName(), context.getVdbVersion()); MatTableEntry matTableEntry = new MatTableEntry(); matTableEntry.valid = valid; + matTableEntry.lastUpdate = loadtime; tables.put(key, matTableEntry, null); } @@ -644,5 +689,4 @@ public MetadataRepository getMetadataRepository() { return this.processorDataManager.getMetadataRepository(); } - } Modified: branches/7.4.x/engine/src/main/java/org/teiid/query/tempdata/TempTableStore.java =================================================================== --- branches/7.4.x/engine/src/main/java/org/teiid/query/tempdata/TempTableStore.java 2011-07-27 21:13:25 UTC (rev 3344) +++ branches/7.4.x/engine/src/main/java/org/teiid/query/tempdata/TempTableStore.java 2011-07-27 21:30:01 UTC (rev 3345) @@ -264,6 +264,7 @@ id = tempMetadataStore.addTempGroup(matTableName, ResolverUtil.resolveElementsInGroup(group, metadata), false, true); id.setQueryNode(metadata.getVirtualPlan(viewId)); id.setCardinality(metadata.getCardinality(viewId)); + id.setOriginalMetadataID(viewId); Object pk = metadata.getPrimaryKey(viewId); if (pk != null) { Modified: branches/7.4.x/engine/src/main/resources/org/teiid/query/i18n.properties =================================================================== --- branches/7.4.x/engine/src/main/resources/org/teiid/query/i18n.properties 2011-07-27 21:13:25 UTC (rev 3344) +++ branches/7.4.x/engine/src/main/resources/org/teiid/query/i18n.properties 2011-07-27 21:30:01 UTC (rev 3345) @@ -813,6 +813,7 @@ TempTableDataManager.failed_load=Failed to load materialized view table {0}. TempTableDataManager.loaded=Loaded materialized view table {0} with row count {1}. TempTableDataManager.loading=Loading materialized view table {0} +TempTableDataManager.cache_load=Loaded materialized view table {0} from cached contents from another clustered node. TempTableDataManager.not_implicit_matview={0} does not target an internal materialized view. TempTableDataManager.row_refresh_pk=Materialized view {0} cannot have a row refreshed since there is no primary key. TempTableDataManager.row_refresh_composite=Materialized view {0} cannot have a row refreshed because it uses a composite key. Modified: branches/7.4.x/jboss-integration/src/main/java/org/teiid/jboss/deployers/RuntimeEngineDeployer.java =================================================================== --- branches/7.4.x/jboss-integration/src/main/java/org/teiid/jboss/deployers/RuntimeEngineDeployer.java 2011-07-27 21:13:25 UTC (rev 3344) +++ branches/7.4.x/jboss-integration/src/main/java/org/teiid/jboss/deployers/RuntimeEngineDeployer.java 2011-07-27 21:30:01 UTC (rev 3345) @@ -113,6 +113,8 @@ import org.teiid.metadata.Table.TriggerEvent; import org.teiid.net.TeiidURL; import org.teiid.query.QueryPlugin; +import org.teiid.query.metadata.QueryMetadataInterface; +import org.teiid.query.metadata.TempMetadataAdapter; import org.teiid.query.metadata.TransformationMetadata; import org.teiid.query.optimizer.relational.RelationalPlanner; import org.teiid.query.processor.DdlPlan; @@ -548,21 +550,11 @@ @Override @ManagementOperation(description="Execute a sql query", params={@ManagementParameter(name="vdbName"),@ManagementParameter(name="vdbVersion"), @ManagementParameter(name="command"), @ManagementParameter(name="timoutInMilli")}) public List<List> executeQuery(final String vdbName, final int version, final String command, final long timoutInMilli) throws AdminException { - Properties properties = new Properties(); - properties.setProperty(TeiidURL.JDBC.VDB_NAME, vdbName); - properties.setProperty(TeiidURL.JDBC.VDB_VERSION, String.valueOf(version)); String user = "JOPR ADMIN"; //$NON-NLS-1$ LogManager.logDetail(LogConstants.CTX_RUNTIME, IntegrationPlugin.Util.getString("admin_executing", user, command)); //$NON-NLS-1$ - SessionMetadata session = null; - try { - session = this.sessionService.createSession(user, null, "JOPR", properties, false, false); //$NON-NLS-1$ - } catch (SessionServiceException e1) { - throw new AdminProcessingException(e1); - } catch (LoginException e1) { - throw new AdminProcessingException(e1); - } + SessionMetadata session = createTemporarySession(vdbName, version, user); final long requestID = 0L; @@ -580,8 +572,12 @@ request.setExecutionId(0L); request.setRowLimit(getMaxRowsFetchSize()); // this would limit the number of rows that are returned. Future<ResultsMessage> message = dqpCore.executeRequest(requestID, request); - ResultsMessage rm = message.get(timoutInMilli, TimeUnit.MILLISECONDS); - + ResultsMessage rm = null; + if (timoutInMilli < 0) { + rm = message.get(); + } else { + rm = message.get(timoutInMilli, TimeUnit.MILLISECONDS); + } if (rm.getException() != null) { throw new AdminProcessingException(rm.getException()); } @@ -616,6 +612,23 @@ } catch (InvalidSessionException e) { //ignore } } + } + + private SessionMetadata createTemporarySession(final String vdbName, final int version, final String user) + throws AdminProcessingException { + Properties properties = new Properties(); + properties.setProperty(TeiidURL.JDBC.VDB_NAME, vdbName); + properties.setProperty(TeiidURL.JDBC.VDB_VERSION, String.valueOf(version)); + + SessionMetadata session = null; + try { + session = this.sessionService.createSession(user, null, "JOPR", properties, false, false); //$NON-NLS-1$ + } catch (SessionServiceException e1) { + throw new AdminProcessingException(e1); + } catch (LoginException e1) { + throw new AdminProcessingException(e1); + } + return session; } /** @@ -685,6 +698,54 @@ } @Override + public void refreshMatView(final String vdbName, final int vdbVersion, final String viewName) { + final VDBMetaData vdb = this.vdbRepository.getVDB(vdbName, vdbVersion); + if (vdb == null) { + return; + } + final TempTableStore globalStore = vdb.getAttachment(TempTableStore.class); + if (globalStore == null) { + return; + } + + Runnable refreshWork = new Runnable() { + @Override + public void run() { + SessionMetadata session = null; + try { + session = createTemporarySession(vdbName, vdbVersion, "internal"); //$NON-NLS-1$ + + DQPWorkContext context = new DQPWorkContext(); + context.setSession(session); + Runnable work = new Runnable() { + @Override + public void run() { + QueryMetadataInterface metadata = vdb.getAttachment(QueryMetadataInterface.class); + TempTableStore tempStore = new TempTableStore("internal"); //$NON-NLS-1$ + TempMetadataAdapter tma = new TempMetadataAdapter(metadata, tempStore.getMetadataStore()); + try { + dqpCore.getDataTierManager().refreshMatView(vdbName, vdbVersion, viewName, tma, globalStore); + } catch (TeiidException e) { + LogManager.logError(LogConstants.CTX_RUNTIME, e, IntegrationPlugin.Util.getString("error_refresh", viewName )); //$NON-NLS-1$ + } + } + }; + context.runInContext(work); + } catch (AdminProcessingException e) { + LogManager.logError(LogConstants.CTX_RUNTIME, e, IntegrationPlugin.Util.getString("error_refresh", viewName )); //$NON-NLS-1$ + } finally { + try { + sessionService.closeSession(session.getSessionId()); + } catch (InvalidSessionException e) { + // ignore + } + } + } + }; + this.dqpCore.addWork(refreshWork); + } + + @Override public void dataModification(String vdbName, int vdbVersion, String schema, String... tableNames) { updateModified(true, vdbName, vdbVersion, schema, tableNames); Modified: branches/7.4.x/jboss-integration/src/main/resources/org/teiid/jboss/i18n.properties =================================================================== --- branches/7.4.x/jboss-integration/src/main/resources/org/teiid/jboss/i18n.properties 2011-07-27 21:13:25 UTC (rev 3344) +++ branches/7.4.x/jboss-integration/src/main/resources/org/teiid/jboss/i18n.properties 2011-07-27 21:30:01 UTC (rev 3345) @@ -48,3 +48,4 @@ admin_executing=JOPR admin {0} is executing command {1} DQPCore.unable_to_process_event=Unable to process event. +error_refresh=error occurred during refreshing the materialized view entries for view {0}