Author: shawkins Date: 2012-06-27 15:35:38 -0400 (Wed, 27 Jun 2012) New Revision: 4209 Added: trunk/engine/src/main/java/org/teiid/query/processor/RegisterRequestParameter.java trunk/engine/src/test/java/org/teiid/query/processor/TestCommonTableOptimizations.java Modified: trunk/api/src/main/java/org/teiid/translator/DataNotAvailableException.java trunk/build/pom.xml trunk/common-core/src/main/java/org/teiid/core/types/basic/StringToByteTransform.java trunk/connectors/sandbox/translator-yahoo/src/main/java/org/teiid/translator/yahoo/TickerCollectorVisitor.java trunk/connectors/sandbox/translator-yahoo/src/main/java/org/teiid/translator/yahoo/YahooExecution.java trunk/connectors/sandbox/translator-yahoo/src/main/java/org/teiid/translator/yahoo/YahooExecutionFactory.java trunk/engine/src/main/java/org/teiid/dqp/internal/process/DataTierManagerImpl.java trunk/engine/src/main/java/org/teiid/query/analysis/AnalysisRecord.java trunk/engine/src/main/java/org/teiid/query/optimizer/relational/PlanToProcessConverter.java trunk/engine/src/main/java/org/teiid/query/processor/ProcessorDataManager.java trunk/engine/src/main/java/org/teiid/query/processor/QueryProcessor.java trunk/engine/src/main/java/org/teiid/query/processor/proc/ExecDynamicSqlInstruction.java trunk/engine/src/main/java/org/teiid/query/processor/proc/ProcedurePlan.java trunk/engine/src/main/java/org/teiid/query/processor/relational/AccessNode.java trunk/engine/src/main/java/org/teiid/query/processor/relational/BatchedUpdateNode.java trunk/engine/src/main/java/org/teiid/query/processor/relational/ProjectIntoNode.java trunk/engine/src/main/java/org/teiid/query/processor/relational/RelationalPlan.java trunk/engine/src/main/java/org/teiid/query/processor/xml/RelationalPlanExecutor.java trunk/engine/src/main/java/org/teiid/query/sql/visitor/EvaluatableVisitor.java trunk/engine/src/main/java/org/teiid/query/tempdata/TempTableDataManager.java trunk/engine/src/main/java/org/teiid/query/util/CommandContext.java trunk/engine/src/test/java/org/teiid/dqp/internal/process/TestDQPCore.java trunk/engine/src/test/java/org/teiid/dqp/internal/process/multisource/TestMultiSourcePlanToProcessConverter.java trunk/engine/src/test/java/org/teiid/query/processor/FakeDataManager.java trunk/engine/src/test/java/org/teiid/query/processor/HardcodedDataManager.java trunk/engine/src/test/java/org/teiid/query/processor/TestProcedureRelational.java trunk/engine/src/test/java/org/teiid/query/processor/TestSourceHints.java trunk/engine/src/test/java/org/teiid/query/processor/relational/TestBatchedUpdateNode.java trunk/engine/src/test/java/org/teiid/query/processor/relational/TestProjectIntoNode.java Log: TEIID-2077 adding intra-query reuse of source queries Modified: trunk/api/src/main/java/org/teiid/translator/DataNotAvailableException.java =================================================================== --- trunk/api/src/main/java/org/teiid/translator/DataNotAvailableException.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/api/src/main/java/org/teiid/translator/DataNotAvailableException.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -26,6 +26,10 @@ /** * Used by asynch connectors to indicate data is not available * and results should be polled for after the given delay in milliseconds. + * <br> + * Note that delays are not guaranteed. The delay is the maximum amount of time before the plan will be re-queued for execution. + * There are several scenarios that would cause the delay to be shorter, such as multiple sources where one source returns a shorter + * delay or if the engine believes more work is to be done before allowing the plan to sit idle. */ public class DataNotAvailableException extends TeiidRuntimeException { Modified: trunk/build/pom.xml =================================================================== --- trunk/build/pom.xml 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/build/pom.xml 2012-06-27 19:35:38 UTC (rev 4209) @@ -80,7 +80,33 @@ </goals> </execution> </executions> - </plugin> + </plugin> + + <plugin> + <groupId>org.codehaus.mojo</groupId> + <artifactId>exec-maven-plugin</artifactId> + <version>1.1.1</version> + <executions> + <execution> + <phase>package</phase> + <goals> + <goal>java</goal> + </goals> + <configuration> + <mainClass>net.sf.retrotranslator.transformer.Retrotranslator</mainClass> + <arguments> + <argument>-srcjar</argument> + <argument>${project.basedir}/target/teiid-${project.version}-jdbc.jar</argument> + <argument>-destjar</argument> + <argument>${project.basedir}/target/teiid-${project.version}-jdbc-jdk15.jar</argument> + <argument>-embed</argument> + <argument>org.teiid.retroruntime</argument> + </arguments> + </configuration> + </execution> + </executions> + </plugin> + </plugins> </build> @@ -105,32 +131,8 @@ </configuration> </plugin> - <plugin> - <groupId>org.codehaus.mojo</groupId> - <artifactId>exec-maven-plugin</artifactId> - <version>1.1.1</version> - <executions> - <execution> - <phase>package</phase> - <goals> - <goal>java</goal> - </goals> - <configuration> - <mainClass>net.sf.retrotranslator.transformer.Retrotranslator</mainClass> - <arguments> - <argument>-srcjar</argument> - <argument>${project.basedir}/target/teiid-${project.version}-jdbc.jar</argument> - <argument>-destjar</argument> - <argument>${project.basedir}/target/teiid-${project.version}-jdbc-jdk15.jar</argument> - <argument>-embed</argument> - <argument>org.teiid.retroruntime</argument> - </arguments> - </configuration> - </execution> - </executions> - </plugin> </plugins> </build> </profile> </profiles> -</project> \ No newline at end of file +</project> Modified: trunk/common-core/src/main/java/org/teiid/core/types/basic/StringToByteTransform.java =================================================================== --- trunk/common-core/src/main/java/org/teiid/core/types/basic/StringToByteTransform.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/common-core/src/main/java/org/teiid/core/types/basic/StringToByteTransform.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -48,7 +48,7 @@ * Type of the incoming value. * @return Source type */ - public Class getSourceType() { + public Class<?> getSourceType() { return String.class; } @@ -56,7 +56,7 @@ * Type of the outgoing value. * @return Target type */ - public Class getTargetType() { + public Class<?> getTargetType() { return Byte.class; } Modified: trunk/connectors/sandbox/translator-yahoo/src/main/java/org/teiid/translator/yahoo/TickerCollectorVisitor.java =================================================================== --- trunk/connectors/sandbox/translator-yahoo/src/main/java/org/teiid/translator/yahoo/TickerCollectorVisitor.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/connectors/sandbox/translator-yahoo/src/main/java/org/teiid/translator/yahoo/TickerCollectorVisitor.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -22,9 +22,15 @@ package org.teiid.translator.yahoo; -import java.util.*; +import java.util.HashSet; +import java.util.List; +import java.util.Set; -import org.teiid.language.*; +import org.teiid.language.Comparison; +import org.teiid.language.Condition; +import org.teiid.language.Expression; +import org.teiid.language.In; +import org.teiid.language.Literal; import org.teiid.language.visitor.HierarchyVisitor; import org.teiid.translator.TranslatorException; @@ -33,21 +39,10 @@ */ public class TickerCollectorVisitor extends HierarchyVisitor { - private Set tickers = new HashSet(); + private Set<String> tickers = new HashSet<String>(); private TranslatorException exception; - /** - * - */ - public TickerCollectorVisitor() { - super(); - } - - public void reset() { - tickers = new HashSet(); - } - - public Set getTickers() { + public Set<String> getTickers() { return this.tickers; } @@ -61,10 +56,8 @@ } public void visit(In obj) { - List exprs = obj.getRightExpressions(); - Iterator iter = exprs.iterator(); - while(iter.hasNext()) { - Expression expr = (Expression) iter.next(); + List<Expression> exprs = obj.getRightExpressions(); + for (Expression expr : exprs) { addTickerFromExpression(expr); } } @@ -84,8 +77,7 @@ } - - public static Set getTickers(Condition crit) throws TranslatorException { + public static Set<String> getTickers(Condition crit) throws TranslatorException { TickerCollectorVisitor visitor = new TickerCollectorVisitor(); crit.acceptVisitor(visitor); Modified: trunk/connectors/sandbox/translator-yahoo/src/main/java/org/teiid/translator/yahoo/YahooExecution.java =================================================================== --- trunk/connectors/sandbox/translator-yahoo/src/main/java/org/teiid/translator/yahoo/YahooExecution.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/connectors/sandbox/translator-yahoo/src/main/java/org/teiid/translator/yahoo/YahooExecution.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -46,10 +46,9 @@ import org.teiid.logging.LogConstants; import org.teiid.logging.LogManager; import org.teiid.metadata.Column; -import org.teiid.metadata.RuntimeMetadata; -import org.teiid.translator.TranslatorException; import org.teiid.translator.DataNotAvailableException; import org.teiid.translator.ResultSetExecution; +import org.teiid.translator.TranslatorException; /** @@ -63,21 +62,17 @@ private static SimpleDateFormat DATE_FORMAT = new SimpleDateFormat("MM/dd/yyyy"); //$NON-NLS-1$ private static SimpleDateFormat TIME_FORMAT = new SimpleDateFormat("HH:mma"); //$NON-NLS-1$ - // Connector resources - private RuntimeMetadata metadata; private Select command; // Execution state - List results; + Iterator<List<?>> results; int[] neededColumns; - int returnIndex = 0; private Select query; /** * */ - public YahooExecution(Select query, RuntimeMetadata metadata) { - this.metadata = metadata; + public YahooExecution(Select query) { this.query = query; } @@ -90,22 +85,22 @@ String yahooUrl = translateIntoUrl(query); // Execute url to get results - this.results = executeUrl(yahooUrl); + this.results = executeUrl(yahooUrl).iterator(); // Determine needed columns in results - this.neededColumns = getNeededColumns(query.getDerivedColumns(), this.metadata); + this.neededColumns = getNeededColumns(query.getDerivedColumns()); } static String translateIntoUrl(Select query) throws TranslatorException { StringBuffer url = new StringBuffer(); url.append(YahooPlugin.Util.getString("YahooExecution.URL_BEGIN")); //$NON-NLS-1$ - Set tickers = getTickers(query); + Set<String> tickers = getTickers(query); if(tickers.size() == 0) { throw new TranslatorException(YahooPlugin.Util.getString("YahooExecution.No_tickers")); //$NON-NLS-1$ } String urlAppendChar = YahooPlugin.Util.getString("YahooExecution.URL_APPEND_CHAR"); //$NON-NLS-1$ - Iterator tickerIter = tickers.iterator(); + Iterator<String> tickerIter = tickers.iterator(); url.append(tickerIter.next()); while(tickerIter.hasNext()) { url.append(urlAppendChar); @@ -119,7 +114,7 @@ /** * @return */ - static Set getTickers(Select query) throws TranslatorException { + static Set<String> getTickers(Select query) throws TranslatorException { Condition crit = query.getWhere(); if(crit == null) { throw new TranslatorException(YahooPlugin.Util.getString("YahooExecution.Must_have_criteria")); //$NON-NLS-1$ @@ -131,8 +126,8 @@ * @param yahooUrl * @return */ - protected List executeUrl(String yahooUrl) throws TranslatorException { - List rows = new ArrayList(); + protected List<List<?>> executeUrl(String yahooUrl) throws TranslatorException { + List<List<?>> rows = new ArrayList<List<?>>(); InputStreamReader inSR = null; BufferedReader buffReader = null; @@ -167,8 +162,12 @@ } catch(IOException e) { throw new TranslatorException(e, e.getMessage()); } finally { - buffReader = null; - inSR = null; + if (buffReader != null) { + try { + buffReader.close(); + } catch (IOException e) { + } + } } return rows; @@ -178,8 +177,8 @@ * @param line * @return */ - static List parseLine(String line) { - List row = new ArrayList(); + static List<Object> parseLine(String line) { + List<Object> row = new ArrayList<Object>(); StringTokenizer rowToken = new StringTokenizer(line,","); //$NON-NLS-1$ for(int i=0; rowToken.hasMoreTokens(); i++){ String data = rowToken.nextToken(); @@ -196,8 +195,10 @@ } else if(i==2) { if(!data.equals("0")){ //$NON-NLS-1$ try { - Date date = DATE_FORMAT.parse(data); - row.add(new java.sql.Date(date.getTime())); + synchronized (DATE_FORMAT) { + Date date = DATE_FORMAT.parse(data); + row.add(new java.sql.Date(date.getTime())); + } } catch(ParseException e) { Object[] params = new Object[] { data, e.getMessage() }; LogManager.logWarning(LogConstants.CTX_CONNECTOR, YahooPlugin.Util.gs(YahooPlugin.Event.TEIID14001, params)); @@ -209,8 +210,10 @@ } else if(i==3) { if(!data.equals("0")){ //$NON-NLS-1$ try { - Date time = TIME_FORMAT.parse(data); - row.add(new java.sql.Time(time.getTime())); + synchronized (TIME_FORMAT) { + Date time = TIME_FORMAT.parse(data); + row.add(new java.sql.Time(time.getTime())); + } } catch(ParseException e) { Object[] params = new Object[] { data, e.getMessage() }; LogManager.logWarning(LogConstants.CTX_CONNECTOR, YahooPlugin.Util.gs(YahooPlugin.Event.TEIID14002, params)); @@ -231,12 +234,13 @@ /** * @param select * @return + * @throws TranslatorException */ - static int[] getNeededColumns(List<DerivedColumn> select, RuntimeMetadata metadata) throws TranslatorException { + static int[] getNeededColumns(List<DerivedColumn> select) throws TranslatorException { int[] cols = new int[select.size()]; - Iterator iter = select.iterator(); + Iterator<DerivedColumn> iter = select.iterator(); for(int i=0; iter.hasNext(); i++) { - DerivedColumn symbol = (DerivedColumn) iter.next(); + DerivedColumn symbol = iter.next(); Expression expr = symbol.getExpression(); if(expr instanceof ColumnReference) { Column element = ((ColumnReference)expr).getMetadataObject(); @@ -250,22 +254,19 @@ } @Override - public List next() throws TranslatorException, DataNotAvailableException { - if (returnIndex < results.size()) { - List row = (List) results.get(returnIndex++); - return projectRow(row, neededColumns); + public List<?> next() throws TranslatorException, DataNotAvailableException { + if (results.hasNext()) { + return projectRow(results.next(), neededColumns); } - return null; } - /** * @param row * @param neededColumns */ - static List projectRow(List row, int[] neededColumns) { - List output = new ArrayList(neededColumns.length); + static List<Object> projectRow(List<?> row, int[] neededColumns) { + List<Object> output = new ArrayList<Object>(neededColumns.length); for(int i=0; i<neededColumns.length; i++) { output.add(row.get(neededColumns[i]-1)); Modified: trunk/connectors/sandbox/translator-yahoo/src/main/java/org/teiid/translator/yahoo/YahooExecutionFactory.java =================================================================== --- trunk/connectors/sandbox/translator-yahoo/src/main/java/org/teiid/translator/yahoo/YahooExecutionFactory.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/connectors/sandbox/translator-yahoo/src/main/java/org/teiid/translator/yahoo/YahooExecutionFactory.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -23,15 +23,15 @@ package org.teiid.translator.yahoo; import java.util.Arrays; -import java.util.Collections; -import java.util.List; import org.teiid.core.types.DataTypeManager; import org.teiid.language.QueryExpression; import org.teiid.language.Select; +import org.teiid.metadata.Column; import org.teiid.metadata.MetadataFactory; import org.teiid.metadata.RuntimeMetadata; import org.teiid.metadata.Table; +import org.teiid.metadata.Column.SearchType; import org.teiid.translator.ExecutionContext; import org.teiid.translator.ExecutionFactory; import org.teiid.translator.ResultSetExecution; @@ -53,13 +53,9 @@ @Override public ResultSetExecution createResultSetExecution(QueryExpression command, ExecutionContext executionContext, RuntimeMetadata metadata, Object connectionFactory) throws TranslatorException { - return new YahooExecution((Select)command, metadata); + return new YahooExecution((Select)command); } - public List getSupportedFunctions() { - return Collections.EMPTY_LIST; - } - public boolean supportsCompareCriteriaEquals() { return true; } @@ -77,14 +73,22 @@ public void getMetadata(MetadataFactory metadataFactory, Object connection) throws TranslatorException { Table t = metadataFactory.addTable("Stock"); //$NON-NLS-1$ metadataFactory.addColumn("symbol", DataTypeManager.DefaultDataTypes.STRING, t); //$NON-NLS-1$ - metadataFactory.addColumn("last", DataTypeManager.DefaultDataTypes.DOUBLE, t); //$NON-NLS-1$ - metadataFactory.addColumn("date", DataTypeManager.DefaultDataTypes.DATE, t); //$NON-NLS-1$ - metadataFactory.addColumn("time", DataTypeManager.DefaultDataTypes.TIME, t); //$NON-NLS-1$ - metadataFactory.addColumn("change", DataTypeManager.DefaultDataTypes.DOUBLE, t); //$NON-NLS-1$ - metadataFactory.addColumn("open", DataTypeManager.DefaultDataTypes.DOUBLE, t); //$NON-NLS-1$ - metadataFactory.addColumn("high", DataTypeManager.DefaultDataTypes.DOUBLE, t); //$NON-NLS-1$ - metadataFactory.addColumn("low", DataTypeManager.DefaultDataTypes.DOUBLE, t); //$NON-NLS-1$ - metadataFactory.addColumn("volume", DataTypeManager.DefaultDataTypes.BIG_INTEGER, t); //$NON-NLS-1$ + Column c = metadataFactory.addColumn("last", DataTypeManager.DefaultDataTypes.DOUBLE, t); //$NON-NLS-1$ + c.setSearchType(SearchType.Unsearchable); + c = metadataFactory.addColumn("date", DataTypeManager.DefaultDataTypes.DATE, t); //$NON-NLS-1$ + c.setSearchType(SearchType.Unsearchable); + c = metadataFactory.addColumn("time", DataTypeManager.DefaultDataTypes.TIME, t); //$NON-NLS-1$ + c.setSearchType(SearchType.Unsearchable); + c = metadataFactory.addColumn("change", DataTypeManager.DefaultDataTypes.DOUBLE, t); //$NON-NLS-1$ + c.setSearchType(SearchType.Unsearchable); + c = metadataFactory.addColumn("open", DataTypeManager.DefaultDataTypes.DOUBLE, t); //$NON-NLS-1$ + c.setSearchType(SearchType.Unsearchable); + c = metadataFactory.addColumn("high", DataTypeManager.DefaultDataTypes.DOUBLE, t); //$NON-NLS-1$ + c.setSearchType(SearchType.Unsearchable); + c = metadataFactory.addColumn("low", DataTypeManager.DefaultDataTypes.DOUBLE, t); //$NON-NLS-1$ + c.setSearchType(SearchType.Unsearchable); + c = metadataFactory.addColumn("volume", DataTypeManager.DefaultDataTypes.BIG_INTEGER, t); //$NON-NLS-1$ + c.setSearchType(SearchType.Unsearchable); metadataFactory.addAccessPattern("needs_symbol", Arrays.asList("symbol"), t); //$NON-NLS-1$ //$NON-NLS-2$ } Modified: trunk/engine/src/main/java/org/teiid/dqp/internal/process/DataTierManagerImpl.java =================================================================== --- trunk/engine/src/main/java/org/teiid/dqp/internal/process/DataTierManagerImpl.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/engine/src/main/java/org/teiid/dqp/internal/process/DataTierManagerImpl.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -69,6 +69,7 @@ import org.teiid.query.optimizer.relational.RelationalPlanner; import org.teiid.query.processor.CollectionTupleSource; import org.teiid.query.processor.ProcessorDataManager; +import org.teiid.query.processor.RegisterRequestParameter; import org.teiid.query.sql.lang.Command; import org.teiid.query.sql.lang.Query; import org.teiid.query.sql.lang.SourceHint; @@ -163,22 +164,22 @@ return eventDistributor; } - public TupleSource registerRequest(CommandContext context, Command command, String modelName, String connectorBindingId, int nodeID, int limit) throws TeiidComponentException, TeiidProcessingException { + public TupleSource registerRequest(CommandContext context, Command command, String modelName, RegisterRequestParameter parameterObject) throws TeiidComponentException, TeiidProcessingException { RequestWorkItem workItem = requestMgr.getRequestWorkItem((RequestID)context.getProcessorID()); if(CoreConstants.SYSTEM_MODEL.equals(modelName) || CoreConstants.SYSTEM_ADMIN_MODEL.equals(modelName)) { return processSystemQuery(context, command, workItem.getDqpWorkContext()); } - AtomicRequestMessage aqr = createRequest(workItem, command, modelName, connectorBindingId, nodeID); + AtomicRequestMessage aqr = createRequest(workItem, command, modelName, parameterObject.connectorBindingId, parameterObject.nodeID); aqr.setCommandContext(context); SourceHint sh = context.getSourceHint(); if (sh != null) { aqr.setGeneralHint(sh.getGeneralHint()); aqr.setHint(sh.getSourceHint(aqr.getConnectorName())); } - if (limit > 0) { - aqr.setFetchSize(Math.min(limit, aqr.getFetchSize())); + if (parameterObject.limit > 0) { + aqr.setFetchSize(Math.min(parameterObject.limit, aqr.getFetchSize())); } if (context.getDataObjects() != null) { for (GroupSymbol gs : GroupCollectorVisitor.getGroupsIgnoreInlineViews(command, false)) { @@ -188,7 +189,7 @@ ConnectorManagerRepository cmr = workItem.getDqpWorkContext().getVDB().getAttachment(ConnectorManagerRepository.class); ConnectorWork work = cmr.getConnectorManager(aqr.getConnectorName()).registerRequest(aqr); work.setRequestWorkItem(workItem); - return new DataTierTupleSource(aqr, workItem, work, this, limit); + return new DataTierTupleSource(aqr, workItem, work, this, parameterObject.limit); } /** Modified: trunk/engine/src/main/java/org/teiid/query/analysis/AnalysisRecord.java =================================================================== --- trunk/engine/src/main/java/org/teiid/query/analysis/AnalysisRecord.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/engine/src/main/java/org/teiid/query/analysis/AnalysisRecord.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -68,6 +68,7 @@ public static final String PROP_GROUP_COLS = "Grouping Columns"; //$NON-NLS-1$ public static final String PROP_SQL = "Query"; //$NON-NLS-1$ public static final String PROP_MODEL_NAME = "Model Name"; //$NON-NLS-1$ + public static final String PROP_SHARING_ID = "Sharing ID"; //$NON-NLS-1$ public static final String PROP_DEPENDENT = "Dependent Join"; //$NON-NLS-1$ public static final String PROP_JOIN_STRATEGY = "Join Strategy"; //$NON-NLS-1$ public static final String PROP_JOIN_TYPE = "Join Type"; //$NON-NLS-1$ Modified: trunk/engine/src/main/java/org/teiid/query/optimizer/relational/PlanToProcessConverter.java =================================================================== --- trunk/engine/src/main/java/org/teiid/query/optimizer/relational/PlanToProcessConverter.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/engine/src/main/java/org/teiid/query/optimizer/relational/PlanToProcessConverter.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -24,6 +24,7 @@ import java.util.ArrayList; import java.util.Collection; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Set; @@ -34,6 +35,7 @@ import org.teiid.core.TeiidComponentException; import org.teiid.core.id.IDGenerator; import org.teiid.core.util.Assertion; +import org.teiid.metadata.FunctionMethod.Determinism; import org.teiid.query.QueryPlugin; import org.teiid.query.analysis.AnalysisRecord; import org.teiid.query.metadata.QueryMetadataInterface; @@ -49,6 +51,7 @@ import org.teiid.query.optimizer.relational.rules.RuleAssignOutputElements; import org.teiid.query.optimizer.relational.rules.RuleChooseJoinStrategy; import org.teiid.query.processor.ProcessorPlan; +import org.teiid.query.processor.RegisterRequestParameter; import org.teiid.query.processor.relational.*; import org.teiid.query.processor.relational.JoinNode.JoinStrategyType; import org.teiid.query.processor.relational.MergeJoinStrategy.SortOption; @@ -61,10 +64,13 @@ import org.teiid.query.sql.symbol.Expression; import org.teiid.query.sql.symbol.ExpressionSymbol; import org.teiid.query.sql.symbol.GroupSymbol; +import org.teiid.query.sql.symbol.Reference; import org.teiid.query.sql.symbol.WindowFunction; import org.teiid.query.sql.util.SymbolMap; import org.teiid.query.sql.visitor.EvaluatableVisitor; import org.teiid.query.sql.visitor.GroupCollectorVisitor; +import org.teiid.query.sql.visitor.ReferenceCollectorVisitor; +import org.teiid.query.sql.visitor.EvaluatableVisitor.EvaluationLevel; public class PlanToProcessConverter { @@ -73,6 +79,15 @@ private AnalysisRecord analysisRecord; private CapabilitiesFinder capFinder; + private Map<List<Object>, AccessNode> sharedCommands = new HashMap<List<Object>, AccessNode>(); + private Map<List<Object>, Integer> topCount = new HashMap<List<Object>, Integer>(); + private int sharedId; + + public static class SharedStateKey { + int id; + int expectedReaders; + } + public PlanToProcessConverter(QueryMetadataInterface metadata, IDGenerator idGenerator, AnalysisRecord analysisRecord, CapabilitiesFinder capFinder) { this.metadata = metadata; this.idGenerator = idGenerator; @@ -82,23 +97,27 @@ public RelationalPlan convert(PlanNode planNode) throws QueryPlannerException, TeiidComponentException { - - boolean debug = analysisRecord.recordDebug(); - if(debug) { - analysisRecord.println("\n============================================================================"); //$NON-NLS-1$ - analysisRecord.println("CONVERTING PLAN TREE TO PROCESS TREE"); //$NON-NLS-1$ - } - - // Convert plan tree nodes into process tree nodes - RelationalNode processNode = convertPlan(planNode); - if(debug) { - analysisRecord.println("\nPROCESS PLAN = \n" + processNode); //$NON-NLS-1$ - analysisRecord.println("============================================================================"); //$NON-NLS-1$ - } - - RelationalPlan processPlan = new RelationalPlan(processNode); - return processPlan; - + try { + boolean debug = analysisRecord.recordDebug(); + if(debug) { + analysisRecord.println("\n============================================================================"); //$NON-NLS-1$ + analysisRecord.println("CONVERTING PLAN TREE TO PROCESS TREE"); //$NON-NLS-1$ + } + + // Convert plan tree nodes into process tree nodes + RelationalNode processNode = convertPlan(planNode); + if(debug) { + analysisRecord.println("\nPROCESS PLAN = \n" + processNode); //$NON-NLS-1$ + analysisRecord.println("============================================================================"); //$NON-NLS-1$ + } + + RelationalPlan processPlan = new RelationalPlan(processNode); + return processPlan; + } finally { + sharedCommands.clear(); + topCount.clear(); + sharedId = 0; + } } private RelationalNode convertPlan(PlanNode planNode) @@ -273,7 +292,7 @@ AccessNode aNode = null; Command command = (Command) node.getProperty(NodeConstants.Info.ATOMIC_REQUEST); Object modelID = node.getProperty(NodeConstants.Info.MODEL_ID); - + EvaluatableVisitor ev = null; if(node.hasBooleanProperty(NodeConstants.Info.IS_DEPENDENT_SET)) { if (command instanceof StoredProcedure) { List references = (List)node.getProperty(NodeConstants.Info.PROCEDURE_INPUTS); @@ -310,7 +329,8 @@ } catch (QueryMetadataException err) { throw new TeiidComponentException(QueryPlugin.Event.TEIID30248, err); } - aNode.setShouldEvaluateExpressions(EvaluatableVisitor.needsProcessingEvaluation(command)); + ev = EvaluatableVisitor.needsEvaluation(command); + aNode.setShouldEvaluateExpressions(ev.requiresEvaluation(EvaluationLevel.PROCESSING)); } if (command instanceof QueryCommand) { @@ -330,6 +350,45 @@ aNode.minimizeProject(command); } setRoutingName(aNode, node); + //check if valid to share this with other nodes + if (ev != null && ev.getDeterminismLevel().compareTo(Determinism.COMMAND_DETERMINISTIC) >= 0 && command.returnsResultSet()) { + //create a top level key to avoid the full command toString + String modelName = aNode.getModelName(); + Collection<GroupSymbol> groups = GroupCollectorVisitor.getGroups(command, true); + List<Object> topKey = new ArrayList<Object>(groups.size() + 1); + topKey.add(modelName); + for (GroupSymbol groupSymbol : groups) { + topKey.add(groupSymbol.toString()); + } + + AccessNode other = sharedCommands.get(topKey); + if (other == null) { + sharedCommands.put(topKey, aNode); + topCount.put(topKey, 1); + } else { + int count = topCount.get(topKey); + if (count == 1) { + Command c = other.getCommand(); + List<Object> key = getCommandKey(c); + sharedCommands.put(key, other); + } + topCount.put(topKey, ++count); + Command c = aNode.getCommand(); + List<Object> key = getCommandKey(c); + + AccessNode initial = this.sharedCommands.get(key); + if (initial != null) { + if (initial.info == null) { + initial.info = new RegisterRequestParameter.SharedAccessInfo(); + initial.info.id = sharedId++; + } + initial.info.sharingCount++; + aNode.info = initial.info; + } else { + this.sharedCommands.put(key, aNode); + } + } + } } break; @@ -491,6 +550,14 @@ return processNode; } + private List<Object> getCommandKey(Command c) { + List<Reference> refs = ReferenceCollectorVisitor.getReferences(c); + List<Object> key = new ArrayList<Object>(2); + key.add(c.toString()); + key.add(refs); + return key; + } + private void updateGroupName(PlanNode node, TableFunctionReference tt) { String groupName = node.getGroups().iterator().next().getName(); tt.getGroupSymbol().setName(groupName); Modified: trunk/engine/src/main/java/org/teiid/query/processor/ProcessorDataManager.java =================================================================== --- trunk/engine/src/main/java/org/teiid/query/processor/ProcessorDataManager.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/engine/src/main/java/org/teiid/query/processor/ProcessorDataManager.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -33,7 +33,7 @@ public interface ProcessorDataManager { - TupleSource registerRequest(CommandContext context, Command command, String modelName, String connectorBindingId, int nodeID, int limit) + TupleSource registerRequest(CommandContext context, Command command, String modelName, RegisterRequestParameter parameterObject) throws TeiidComponentException, TeiidProcessingException; /** Modified: trunk/engine/src/main/java/org/teiid/query/processor/QueryProcessor.java =================================================================== --- trunk/engine/src/main/java/org/teiid/query/processor/QueryProcessor.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/engine/src/main/java/org/teiid/query/processor/QueryProcessor.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -22,30 +22,36 @@ package org.teiid.query.processor; +import java.util.HashMap; import java.util.List; +import java.util.Map; import org.teiid.common.buffer.BlockedException; import org.teiid.common.buffer.BufferManager; import org.teiid.common.buffer.TupleBatch; import org.teiid.common.buffer.TupleBuffer; +import org.teiid.common.buffer.TupleSource; import org.teiid.common.buffer.BufferManager.BufferReserveMode; +import org.teiid.common.buffer.BufferManager.TupleSourceType; import org.teiid.core.TeiidComponentException; import org.teiid.core.TeiidException; import org.teiid.core.TeiidProcessingException; import org.teiid.core.TeiidRuntimeException; import org.teiid.core.util.Assertion; +import org.teiid.events.EventDistributor; import org.teiid.logging.LogConstants; import org.teiid.logging.LogManager; import org.teiid.logging.MessageLevel; import org.teiid.query.QueryPlugin; import org.teiid.query.processor.BatchCollector.BatchProducer; +import org.teiid.query.sql.lang.Command; import org.teiid.query.util.CommandContext; /** * Driver for plan processing. */ -public class QueryProcessor implements BatchProducer { - +public class QueryProcessor implements BatchProducer, ProcessorDataManager { + public static class ExpiredTimeSliceException extends TeiidRuntimeException { private static final long serialVersionUID = 4585044674826578060L; } @@ -56,6 +62,56 @@ QueryProcessor createQueryProcessor(String query, String recursionGroup, CommandContext commandContext, Object... params) throws TeiidProcessingException, TeiidComponentException; } + private class SharedState { + TupleBuffer tb; + TupleSource ts; + int id; + int expectedReaders; + + private void remove() { + ts.closeSource(); + tb.remove(); + tb = null; + ts = null; + } + } + + private final class BufferedTupleSource implements TupleSource { + private int rowNumber = 1; + private SharedState state; + + private BufferedTupleSource(SharedState state) { + this.state = state; + } + + @Override + public List<?> nextTuple() throws TeiidComponentException, + TeiidProcessingException { + if (rowNumber <= state.tb.getRowCount()) { + return state.tb.getBatch(rowNumber).getTuple(rowNumber++); + } + if (state.tb.isFinal()) { + return null; + } + List<?> row = state.ts.nextTuple(); + if (row == null) { + state.tb.setFinal(true); + } else { + this.state.tb.addTuple(row); + rowNumber++; + } + return row; + } + + @Override + public void closeSource() { + if (--state.expectedReaders == 0) { + state.remove(); + sharedStates.remove(state.id); + } + } + } + private CommandContext context; private ProcessorDataManager dataMgr; private BufferManager bufferMgr; @@ -69,6 +125,8 @@ private boolean processorClosed; private boolean continuous; private int rowOffset = 1; + + Map<Integer, SharedState> sharedStates; /** * Construct a processor with all necessary information to process. @@ -78,9 +136,9 @@ * @param dataMgr The data manager that provides access to get data * @throws TeiidComponentException */ - public QueryProcessor(ProcessorPlan plan, CommandContext context, BufferManager bufferMgr, ProcessorDataManager dataMgr) throws TeiidComponentException { + public QueryProcessor(ProcessorPlan plan, CommandContext context, BufferManager bufferMgr, final ProcessorDataManager dataMgr) { this.context = context; - this.dataMgr = dataMgr; + this.dataMgr = dataMgr; this.processPlan = plan; this.bufferMgr = bufferMgr; } @@ -187,7 +245,7 @@ // initialize if necessary if(!initialized) { reserved = this.bufferMgr.reserveBuffers(this.bufferMgr.getSchemaSize(this.getOutputElements()), BufferReserveMode.FORCE); - this.processPlan.initialize(context, this.dataMgr, bufferMgr); + this.processPlan.initialize(context, this, bufferMgr); initialized = true; } @@ -208,6 +266,12 @@ if (LogManager.isMessageToBeRecorded(LogConstants.CTX_DQP, MessageLevel.DETAIL)) { LogManager.logDetail(LogConstants.CTX_DQP, "QueryProcessor: closing processor"); //$NON-NLS-1$ } + if (sharedStates != null) { + for (SharedState ss : sharedStates.values()) { + ss.remove(); + } + sharedStates = null; + } this.bufferMgr.releaseBuffers(reserved); reserved = 0; processorClosed = true; @@ -282,4 +346,40 @@ public void setContinuous(boolean continuous) { this.continuous = continuous; } + + @Override + public Object lookupCodeValue(CommandContext ctx, String codeTableName, + String returnElementName, String keyElementName, Object keyValue) + throws BlockedException, TeiidComponentException, + TeiidProcessingException { + return dataMgr.lookupCodeValue(ctx, codeTableName, returnElementName, keyElementName, keyValue); + } + + @Override + public EventDistributor getEventDistributor() { + return dataMgr.getEventDistributor(); + } + + @Override + public TupleSource registerRequest(CommandContext ctx, Command command, + String modelName, RegisterRequestParameter parameterObject) + throws TeiidComponentException, TeiidProcessingException { + if (parameterObject.info == null) { + return dataMgr.registerRequest(ctx, command, modelName, parameterObject); + } + //begin handling of shared commands + if (sharedStates == null) { + sharedStates = new HashMap<Integer, SharedState>(); + } + SharedState state = sharedStates.get(parameterObject.info.id); + if (state == null) { + state = new SharedState(); + state.expectedReaders = parameterObject.info.sharingCount; + state.tb = QueryProcessor.this.bufferMgr.createTupleBuffer(command.getProjectedSymbols(), ctx.getConnectionId(), TupleSourceType.PROCESSOR); + state.ts = dataMgr.registerRequest(ctx, command, modelName, new RegisterRequestParameter(parameterObject.connectorBindingId, 0, -1)); + state.id = parameterObject.info.id; + sharedStates.put(parameterObject.info.id, state); + } + return new BufferedTupleSource(state); + } } Added: trunk/engine/src/main/java/org/teiid/query/processor/RegisterRequestParameter.java =================================================================== --- trunk/engine/src/main/java/org/teiid/query/processor/RegisterRequestParameter.java (rev 0) +++ trunk/engine/src/main/java/org/teiid/query/processor/RegisterRequestParameter.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -0,0 +1,46 @@ +/* + * 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.query.processor; + +public class RegisterRequestParameter { + + public static class SharedAccessInfo { + public int id; + public int sharingCount = 1; + } + + public String connectorBindingId; + public int nodeID = 0; + public int limit = -1; + public SharedAccessInfo info; + + public RegisterRequestParameter(String connectorBindingId, int nodeID, + int limit) { + this.connectorBindingId = connectorBindingId; + this.nodeID = nodeID; + this.limit = limit; + } + + public RegisterRequestParameter() { + } +} \ No newline at end of file Property changes on: trunk/engine/src/main/java/org/teiid/query/processor/RegisterRequestParameter.java ___________________________________________________________________ Added: svn:mime-type + text/plain Modified: trunk/engine/src/main/java/org/teiid/query/processor/proc/ExecDynamicSqlInstruction.java =================================================================== --- trunk/engine/src/main/java/org/teiid/query/processor/proc/ExecDynamicSqlInstruction.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/engine/src/main/java/org/teiid/query/processor/proc/ExecDynamicSqlInstruction.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -51,6 +51,7 @@ import org.teiid.query.optimizer.capabilities.CapabilitiesFinder; import org.teiid.query.parser.QueryParser; import org.teiid.query.processor.ProcessorPlan; +import org.teiid.query.processor.RegisterRequestParameter; import org.teiid.query.resolver.QueryResolver; import org.teiid.query.rewriter.QueryRewriter; import org.teiid.query.sql.ProcedureReservedWords; @@ -209,7 +210,7 @@ c.setRuntimeType(DataTypeManager.getDataTypeName(es.getType())); create.getColumns().add(c); } - procEnv.getDataManager().registerRequest(procEnv.getContext(), create, TempMetadataAdapter.TEMP_MODEL.getName(), null, 0, -1); + procEnv.getDataManager().registerRequest(procEnv.getContext(), create, TempMetadataAdapter.TEMP_MODEL.getName(), new RegisterRequestParameter()); } } Modified: trunk/engine/src/main/java/org/teiid/query/processor/proc/ProcedurePlan.java =================================================================== --- trunk/engine/src/main/java/org/teiid/query/processor/proc/ProcedurePlan.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/engine/src/main/java/org/teiid/query/processor/proc/ProcedurePlan.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -66,6 +66,7 @@ import org.teiid.query.processor.ProcessorDataManager; import org.teiid.query.processor.ProcessorPlan; import org.teiid.query.processor.QueryProcessor; +import org.teiid.query.processor.RegisterRequestParameter; import org.teiid.query.processor.relational.SubqueryAwareEvaluator; import org.teiid.query.sql.ProcedureReservedWords; import org.teiid.query.sql.lang.Command; @@ -161,9 +162,9 @@ @Override public TupleSource registerRequest(CommandContext context, Command command, - String modelName, String connectorBindingId, int nodeID, int limit) + String modelName, RegisterRequestParameter parameterObject) throws TeiidComponentException, TeiidProcessingException { - TupleSource ts = parentDataMrg.registerRequest(context, command, modelName, connectorBindingId, nodeID, limit); + TupleSource ts = parentDataMrg.registerRequest(context, command, modelName, parameterObject); if (blockContext != null && ts instanceof DataTierTupleSource) { txnTupleSources.add(new WeakReference<DataTierTupleSource>((DataTierTupleSource)ts)); } Modified: trunk/engine/src/main/java/org/teiid/query/processor/relational/AccessNode.java =================================================================== --- trunk/engine/src/main/java/org/teiid/query/processor/relational/AccessNode.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/engine/src/main/java/org/teiid/query/processor/relational/AccessNode.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -43,6 +43,7 @@ import org.teiid.query.eval.Evaluator; import org.teiid.query.metadata.QueryMetadataInterface; import org.teiid.query.processor.ProcessorDataManager; +import org.teiid.query.processor.RegisterRequestParameter; import org.teiid.query.rewriter.QueryRewriter; import org.teiid.query.sql.lang.Command; import org.teiid.query.sql.lang.OrderByItem; @@ -74,6 +75,8 @@ private Object[] projection; private List<Expression> originalSelect; private Object modelId; + + public RegisterRequestParameter.SharedAccessInfo info; protected AccessNode() { super(); @@ -328,7 +331,7 @@ } @Override - protected void addBatchRow(List row) { + protected void addBatchRow(List<?> row) { if (this.getOutputElements().isEmpty()) { //a dummy column was added to the query, just remove it now row = Collections.emptyList(); @@ -362,7 +365,9 @@ limit = parent.getLimit() + parent.getOffset(); } } - tupleSources.add(getDataManager().registerRequest(getContext(), atomicCommand, modelName, connectorBindingId, getID(), limit)); + RegisterRequestParameter param = new RegisterRequestParameter(connectorBindingId, getID(), limit); + param.info = info; + tupleSources.add(getDataManager().registerRequest(getContext(), atomicCommand, modelName, param)); if (tupleSources.size() > 1) { reserved += getBufferManager().reserveBuffers(schemaSize, BufferReserveMode.FORCE); } @@ -393,6 +398,9 @@ protected void getNodeString(StringBuffer str) { super.getNodeString(str); str.append(command); + if (this.info != null) { + str.append(" [SHARED ").append(this.info.id).append("]"); //$NON-NLS-1$ //$NON-NLS-2$ + } } public Object clone(){ @@ -412,12 +420,16 @@ target.originalSelect = source.originalSelect; } target.command = source.command; + target.info = source.info; } public PlanNode getDescriptionProperties() { PlanNode props = super.getDescriptionProperties(); props.addProperty(PROP_SQL, this.command.toString()); props.addProperty(PROP_MODEL_NAME, this.modelName); + if (this.info != null) { + props.addProperty(PROP_SHARING_ID, String.valueOf(this.info.id)); + } return props; } Modified: trunk/engine/src/main/java/org/teiid/query/processor/relational/BatchedUpdateNode.java =================================================================== --- trunk/engine/src/main/java/org/teiid/query/processor/relational/BatchedUpdateNode.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/engine/src/main/java/org/teiid/query/processor/relational/BatchedUpdateNode.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -36,6 +36,7 @@ import org.teiid.core.TeiidProcessingException; import org.teiid.query.QueryPlugin; import org.teiid.query.eval.Evaluator; +import org.teiid.query.processor.RegisterRequestParameter; import org.teiid.query.sql.lang.BatchedUpdateCommand; import org.teiid.query.sql.lang.Command; import org.teiid.query.sql.util.VariableContext; @@ -113,7 +114,7 @@ } if (!commandsToExecute.isEmpty()) { BatchedUpdateCommand command = new BatchedUpdateCommand(commandsToExecute); - tupleSource = getDataManager().registerRequest(getContext(), command, modelName, null, getID(), -1); + tupleSource = getDataManager().registerRequest(getContext(), command, modelName, new RegisterRequestParameter(null, getID(), -1)); } } Modified: trunk/engine/src/main/java/org/teiid/query/processor/relational/ProjectIntoNode.java =================================================================== --- trunk/engine/src/main/java/org/teiid/query/processor/relational/ProjectIntoNode.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/engine/src/main/java/org/teiid/query/processor/relational/ProjectIntoNode.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -38,6 +38,7 @@ import org.teiid.common.buffer.BufferManager.TupleSourceType; import org.teiid.core.TeiidComponentException; import org.teiid.core.TeiidProcessingException; +import org.teiid.query.processor.RegisterRequestParameter; import org.teiid.query.sql.lang.BatchedUpdateCommand; import org.teiid.query.sql.lang.Command; import org.teiid.query.sql.lang.Insert; @@ -209,7 +210,7 @@ } private void registerRequest(Command command) throws TeiidComponentException, TeiidProcessingException { - tupleSource = getDataManager().registerRequest(getContext(), command, this.modelName, null, getID(), -1); + tupleSource = getDataManager().registerRequest(getContext(), command, this.modelName, new RegisterRequestParameter(null, getID(), -1)); } private void closeRequest() { Modified: trunk/engine/src/main/java/org/teiid/query/processor/relational/RelationalPlan.java =================================================================== --- trunk/engine/src/main/java/org/teiid/query/processor/relational/RelationalPlan.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/engine/src/main/java/org/teiid/query/processor/relational/RelationalPlan.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -39,6 +39,7 @@ import org.teiid.query.processor.ProcessorDataManager; import org.teiid.query.processor.ProcessorPlan; import org.teiid.query.processor.QueryProcessor; +import org.teiid.query.processor.RegisterRequestParameter; import org.teiid.query.processor.relational.ProjectIntoNode.Mode; import org.teiid.query.sql.LanguageObject; import org.teiid.query.sql.lang.Create; @@ -148,13 +149,13 @@ Create create = new Create(); create.setElementSymbolsAsColumns(withCommand.getColumns()); create.setTable(withCommand.getGroupSymbol()); - this.root.getDataManager().registerRequest(getContext(), create, TempMetadataAdapter.TEMP_MODEL.getID(), null, 0, -1); + this.root.getDataManager().registerRequest(getContext(), create, TempMetadataAdapter.TEMP_MODEL.getID(), new RegisterRequestParameter()); } while (true) { TupleBatch batch = withProcessor.nextBatch(); Insert insert = new Insert(withCommand.getGroupSymbol(), withCommand.getColumns(), null); insert.setTupleSource(new CollectionTupleSource(batch.getTuples().iterator())); - this.root.getDataManager().registerRequest(getContext(), insert, TempMetadataAdapter.TEMP_MODEL.getID(), null, 0, -1); + this.root.getDataManager().registerRequest(getContext(), insert, TempMetadataAdapter.TEMP_MODEL.getID(), new RegisterRequestParameter()); if (batch.getTerminationFlag()) { break; } Modified: trunk/engine/src/main/java/org/teiid/query/processor/xml/RelationalPlanExecutor.java =================================================================== --- trunk/engine/src/main/java/org/teiid/query/processor/xml/RelationalPlanExecutor.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/engine/src/main/java/org/teiid/query/processor/xml/RelationalPlanExecutor.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -39,6 +39,7 @@ import org.teiid.query.processor.ProcessorDataManager; import org.teiid.query.processor.ProcessorPlan; import org.teiid.query.processor.QueryProcessor; +import org.teiid.query.processor.RegisterRequestParameter; import org.teiid.query.sql.lang.Insert; import org.teiid.query.sql.symbol.ElementSymbol; import org.teiid.query.sql.util.VariableContext; @@ -129,15 +130,15 @@ Insert insert = this.resultInfo.getTempInsert(); insert.setTupleSource(new TempLoadTupleSource()); - this.dataManager.registerRequest(this.internalProcessor.getContext(), insert, TempMetadataAdapter.TEMP_MODEL.getName(), null, 0, -1); + this.dataManager.registerRequest(this.internalProcessor.getContext(), insert, TempMetadataAdapter.TEMP_MODEL.getName(), new RegisterRequestParameter()); if (!doneLoading) { throw BlockedException.block("Blocking on result set load"); //$NON-NLS-1$ } internalProcessor.closeProcessing(); AlterTempTable att = new AlterTempTable(tempTable); //mark the temp table as non-updatable - this.dataManager.registerRequest(this.internalProcessor.getContext(), att, TempMetadataAdapter.TEMP_MODEL.getName(), null, 0, -1); - this.tupleSource = this.dataManager.registerRequest(this.internalProcessor.getContext(), this.resultInfo.getTempSelect(), TempMetadataAdapter.TEMP_MODEL.getName(), null, 0, -1); + this.dataManager.registerRequest(this.internalProcessor.getContext(), att, TempMetadataAdapter.TEMP_MODEL.getName(), new RegisterRequestParameter()); + this.tupleSource = this.dataManager.registerRequest(this.internalProcessor.getContext(), this.resultInfo.getTempSelect(), TempMetadataAdapter.TEMP_MODEL.getName(), new RegisterRequestParameter()); } //force execution currentRow(); @@ -146,7 +147,7 @@ AlterTempTable att = new AlterTempTable(tempTable); //TODO: if the parent is small, then this is not necessary att.setIndexColumns(this.resultInfo.getFkColumns()); - this.dataManager.registerRequest(this.internalProcessor.getContext(), att, TempMetadataAdapter.TEMP_MODEL.getName(), null, 0, -1); + this.dataManager.registerRequest(this.internalProcessor.getContext(), att, TempMetadataAdapter.TEMP_MODEL.getName(), new RegisterRequestParameter()); } this.currentRowNumber = 0; @@ -220,7 +221,7 @@ LogManager.logDetail(LogConstants.CTX_XML_PLAN, "Unloading result set temp table", rsTempTable); //$NON-NLS-1$ internalProcessor.closeProcessing(); try { - this.tupleSource = this.dataManager.registerRequest(this.internalProcessor.getContext(), this.resultInfo.getTempDrop(), TempMetadataAdapter.TEMP_MODEL.getName(), null, 0, -1); + this.tupleSource = this.dataManager.registerRequest(this.internalProcessor.getContext(), this.resultInfo.getTempDrop(), TempMetadataAdapter.TEMP_MODEL.getName(), new RegisterRequestParameter()); } catch (TeiidProcessingException e) { LogManager.logDetail(org.teiid.logging.LogConstants.CTX_XML_PLAN, e, "Error dropping result set temp table", rsTempTable); //$NON-NLS-1$ } Modified: trunk/engine/src/main/java/org/teiid/query/sql/visitor/EvaluatableVisitor.java =================================================================== --- trunk/engine/src/main/java/org/teiid/query/sql/visitor/EvaluatableVisitor.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/engine/src/main/java/org/teiid/query/sql/visitor/EvaluatableVisitor.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -20,9 +20,6 @@ * 02110-1301 USA. */ -/** - * - */ package org.teiid.query.sql.visitor; import java.util.TreeSet; @@ -155,6 +152,8 @@ } public void visit(DependentSetCriteria obj) { + //without knowing what is feeding this, we need to treat it as non-deterministic + setDeterminismLevel(Determinism.NONDETERMINISTIC); evaluationNotPossible(EvaluationLevel.PROCESSING); } @@ -218,4 +217,22 @@ DeepPreOrderNavigator.doVisit(obj, visitor); return visitor.levels.contains(EvaluationLevel.PROCESSING); } + + public boolean requiresEvaluation(EvaluationLevel evaluationLevel) { + return levels.contains(evaluationLevel); + } + + public Determinism getDeterminismLevel() { + return determinismLevel; + } + + public boolean hasCorrelatedReferences() { + return hasCorrelatedReferences; + } + + public static final EvaluatableVisitor needsEvaluation(LanguageObject obj) { + EvaluatableVisitor visitor = new EvaluatableVisitor(); + DeepPreOrderNavigator.doVisit(obj, visitor); + return visitor; + } } Modified: trunk/engine/src/main/java/org/teiid/query/tempdata/TempTableDataManager.java =================================================================== --- trunk/engine/src/main/java/org/teiid/query/tempdata/TempTableDataManager.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/engine/src/main/java/org/teiid/query/tempdata/TempTableDataManager.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -67,6 +67,7 @@ import org.teiid.query.processor.CollectionTupleSource; import org.teiid.query.processor.ProcessorDataManager; import org.teiid.query.processor.QueryProcessor; +import org.teiid.query.processor.RegisterRequestParameter; import org.teiid.query.resolver.util.ResolverUtil; import org.teiid.query.sql.lang.*; import org.teiid.query.sql.navigator.PostOrderNavigator; @@ -122,7 +123,7 @@ CommandContext context, Command command, String modelName, - String connectorBindingId, int nodeID, int limit) + RegisterRequestParameter parameterObject) throws TeiidComponentException, TeiidProcessingException { TempTableStore tempTableStore = context.getTempTableStore(); @@ -132,7 +133,7 @@ return result; } } - return this.processorDataManager.registerRequest(context, command, modelName, connectorBindingId, nodeID, limit); + return this.processorDataManager.registerRequest(context, command, modelName, parameterObject); } TupleSource registerRequest(CommandContext context, String modelName, Command command) throws TeiidComponentException, TeiidProcessingException { Modified: trunk/engine/src/main/java/org/teiid/query/util/CommandContext.java =================================================================== --- trunk/engine/src/main/java/org/teiid/query/util/CommandContext.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/engine/src/main/java/org/teiid/query/util/CommandContext.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -706,7 +706,7 @@ } } - public static DecimalFormat getDecimalFormat(CommandContext context, String format) { + public static DecimalFormat getDecimalFormat(CommandContext context, String format) { DecimalFormat result = null; if (context != null) { if (context.globalState.decimalFormatCache == null) { Modified: trunk/engine/src/test/java/org/teiid/dqp/internal/process/TestDQPCore.java =================================================================== --- trunk/engine/src/test/java/org/teiid/dqp/internal/process/TestDQPCore.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/engine/src/test/java/org/teiid/dqp/internal/process/TestDQPCore.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -59,6 +59,7 @@ import org.teiid.query.optimizer.capabilities.BasicSourceCapabilities; import org.teiid.query.optimizer.capabilities.SourceCapabilities.Capability; import org.teiid.query.unittest.RealMetadataFactory; +import org.teiid.translator.SourceSystemFunctions; @SuppressWarnings("nls") public class TestDQPCore { @@ -397,13 +398,16 @@ @Test public void testSourceConcurrency() throws Exception { //setup default of 2 agds.setSleep(100); + BasicSourceCapabilities bsc = TestOptimizer.getTypicalCapabilities(); + bsc.setFunctionSupport(SourceSystemFunctions.CONCAT, true); + agds.setCaps(bsc); StringBuffer sql = new StringBuffer(); int branches = 20; for (int i = 0; i < branches; i++) { if (i > 0) { sql.append(" union all "); } - sql.append("select intkey || " + i + " from bqt1.smalla"); + sql.append("select stringkey || " + i + " from bqt1.smalla"); } sql.append(" limit 2"); helpExecute(sql.toString(), "a"); Modified: trunk/engine/src/test/java/org/teiid/dqp/internal/process/multisource/TestMultiSourcePlanToProcessConverter.java =================================================================== --- trunk/engine/src/test/java/org/teiid/dqp/internal/process/multisource/TestMultiSourcePlanToProcessConverter.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/engine/src/test/java/org/teiid/dqp/internal/process/multisource/TestMultiSourcePlanToProcessConverter.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -48,6 +48,7 @@ import org.teiid.query.processor.HardcodedDataManager; import org.teiid.query.processor.ProcessorDataManager; import org.teiid.query.processor.ProcessorPlan; +import org.teiid.query.processor.RegisterRequestParameter; import org.teiid.query.processor.TestProcessor; import org.teiid.query.resolver.TestResolver; import org.teiid.query.rewriter.QueryRewriter; @@ -73,8 +74,8 @@ setMustRegisterCommands(false); } - public TupleSource registerRequest(CommandContext context, Command command, String modelName, String connectorBindingId, int nodeID, int limit) throws org.teiid.core.TeiidComponentException { - assertNotNull(connectorBindingId); + public TupleSource registerRequest(CommandContext context, Command command, String modelName, RegisterRequestParameter parameterObject) throws org.teiid.core.TeiidComponentException { + assertNotNull(parameterObject.connectorBindingId); Collection<ElementSymbol> elements = ElementCollectorVisitor.getElements(command, true, true); @@ -83,7 +84,7 @@ fail("Query Contains a MultiSourceElement -- MultiSource expansion did not happen"); //$NON-NLS-1$ } } - return super.registerRequest(context, command, modelName, connectorBindingId, nodeID, limit); + return super.registerRequest(context, command, modelName, parameterObject); } } Modified: trunk/engine/src/test/java/org/teiid/query/processor/FakeDataManager.java =================================================================== --- trunk/engine/src/test/java/org/teiid/query/processor/FakeDataManager.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/engine/src/test/java/org/teiid/query/processor/FakeDataManager.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -103,10 +103,10 @@ procTuples.put(proc, data); } - public TupleSource registerRequest(CommandContext context, Command command, String modelName, String connectorBindingId, int nodeID, int limit) + public TupleSource registerRequest(CommandContext context, Command command, String modelName, RegisterRequestParameter parameterObject) throws TeiidComponentException { - LogManager.logTrace(LOG_CONTEXT, new Object[]{"Register Request:", command, ",processorID:", context.getProcessorID(), ",model name:", modelName,",TupleSourceID nodeID:",new Integer(nodeID)}); //$NON-NLS-1$ //$NON-NLS-2$ //$NON-NLS-3$ //$NON-NLS-4$ + LogManager.logTrace(LOG_CONTEXT, new Object[]{"Register Request:", command, ",processorID:", context.getProcessorID(), ",model name:", modelName,",TupleSourceID nodeID:",new Integer(parameterObject.nodeID)}); //$NON-NLS-1$ //$NON-NLS-2$ //$NON-NLS-3$ //$NON-NLS-4$ if (this.recordingCommands) { if (! (command instanceof BatchedUpdateCommand) ) { Modified: trunk/engine/src/test/java/org/teiid/query/processor/HardcodedDataManager.java =================================================================== --- trunk/engine/src/test/java/org/teiid/query/processor/HardcodedDataManager.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/engine/src/test/java/org/teiid/query/processor/HardcodedDataManager.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -120,13 +120,13 @@ } /** - * @see org.teiid.query.processor.ProcessorDataManager#registerRequest(CommandContext, org.teiid.query.sql.lang.Command, java.lang.String, String, int, int) + * @see org.teiid.query.processor.ProcessorDataManager#registerRequest(CommandContext, org.teiid.query.sql.lang.Command, java.lang.String, RegisterRequestParameter) * @since 4.2 */ public TupleSource registerRequest(CommandContext context, Command command, String modelName, - String connectorBindingId, int nodeID, int limit) throws TeiidComponentException { + RegisterRequestParameter parameterObject) throws TeiidComponentException { if(modelName != null && validModels != null && ! validModels.contains(modelName)) { throw new TeiidComponentException("Detected query against invalid model: " + modelName + ": " + command); //$NON-NLS-1$//$NON-NLS-2$ Added: trunk/engine/src/test/java/org/teiid/query/processor/TestCommonTableOptimizations.java =================================================================== --- trunk/engine/src/test/java/org/teiid/query/processor/TestCommonTableOptimizations.java (rev 0) +++ trunk/engine/src/test/java/org/teiid/query/processor/TestCommonTableOptimizations.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -0,0 +1,56 @@ +/* + * 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.query.processor; + +import static org.junit.Assert.*; +import static org.teiid.query.processor.TestProcessor.*; + +import java.util.Arrays; +import java.util.List; + +import org.junit.Test; +import org.teiid.query.unittest.RealMetadataFactory; + +@SuppressWarnings("nls") +public class TestCommonTableOptimizations { + + @Test public void testDuplicateSourceQuery() { + String sql = "SELECT e1 FROM pm1.g1 union all select e1 from pm1.g1"; //$NON-NLS-1$ + + List<?>[] expected = new List[] { + Arrays.asList("a"), //$NON-NLS-1$ + Arrays.asList("b"), //$NON-NLS-1$ + Arrays.asList("a"), //$NON-NLS-1$ + Arrays.asList("b"), //$NON-NLS-1$ + }; + + HardcodedDataManager dataManager = new HardcodedDataManager(); + dataManager.addData("SELECT pm1.g1.e1 FROM pm1.g1", new List<?>[] {Arrays.asList("a"), Arrays.asList("b")}); + + ProcessorPlan plan = helpGetPlan(sql, RealMetadataFactory.example1Cached()); + + helpProcess(plan, dataManager, expected); + assertEquals(3, dataManager.getCommandHistory().size()); + } + +} Property changes on: trunk/engine/src/test/java/org/teiid/query/processor/TestCommonTableOptimizations.java ___________________________________________________________________ Added: svn:mime-type + text/plain Modified: trunk/engine/src/test/java/org/teiid/query/processor/TestProcedureRelational.java =================================================================== --- trunk/engine/src/test/java/org/teiid/query/processor/TestProcedureRelational.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/engine/src/test/java/org/teiid/query/processor/TestProcedureRelational.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -754,7 +754,7 @@ @Override public TupleSource registerRequest(CommandContext context, Command command, String modelName, - String connectorBindingId, int nodeID, int limit) + RegisterRequestParameter parameterObject) throws TeiidComponentException { if (command instanceof StoredProcedure) { StoredProcedure proc = (StoredProcedure)command; @@ -766,7 +766,7 @@ }); } return super.registerRequest(context, command, modelName, - connectorBindingId, nodeID, limit); + parameterObject); } }; Modified: trunk/engine/src/test/java/org/teiid/query/processor/TestSourceHints.java =================================================================== --- trunk/engine/src/test/java/org/teiid/query/processor/TestSourceHints.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/engine/src/test/java/org/teiid/query/processor/TestSourceHints.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -91,7 +91,7 @@ @Override public TupleSource registerRequest(CommandContext context, Command command, String modelName, - String connectorBindingId, int nodeID, int limit) + RegisterRequestParameter parameterObject) throws TeiidComponentException { if (general == null && hint == null) { assertNull(context.getSourceHint()); Modified: trunk/engine/src/test/java/org/teiid/query/processor/relational/TestBatchedUpdateNode.java =================================================================== --- trunk/engine/src/test/java/org/teiid/query/processor/relational/TestBatchedUpdateNode.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/engine/src/test/java/org/teiid/query/processor/relational/TestBatchedUpdateNode.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -42,6 +42,7 @@ import org.teiid.query.metadata.QueryMetadataInterface; import org.teiid.query.optimizer.TestBatchedUpdatePlanner; import org.teiid.query.processor.ProcessorDataManager; +import org.teiid.query.processor.RegisterRequestParameter; import org.teiid.query.sql.lang.BatchedUpdateCommand; import org.teiid.query.sql.lang.Command; import org.teiid.query.sql.visitor.EvaluatableVisitor; @@ -214,10 +215,10 @@ this.numExecutedCommands = numExecutedCommands; } public Object lookupCodeValue(CommandContext context,String codeTableName,String returnElementName,String keyElementName,Object keyValue) throws BlockedException,TeiidComponentException {return null;} - public TupleSource registerRequest(CommandContext context,Command command,String modelName,String connectorBindingId, int nodeID, int limit) throws TeiidComponentException { + public TupleSource registerRequest(CommandContext context,Command command,String modelName,RegisterRequestParameter parameterObject) throws TeiidComponentException { assertEquals("myProcessorID", context.getProcessorID()); //$NON-NLS-1$ assertEquals("myModelName", modelName); //$NON-NLS-1$ - assertEquals(1, nodeID); + assertEquals(1, parameterObject.nodeID); commands.add(command.toString()); actualCommands.add(command); return new FakeTupleSource(numExecutedCommands); Modified: trunk/engine/src/test/java/org/teiid/query/processor/relational/TestProjectIntoNode.java =================================================================== --- trunk/engine/src/test/java/org/teiid/query/processor/relational/TestProjectIntoNode.java 2012-06-27 17:07:54 UTC (rev 4208) +++ trunk/engine/src/test/java/org/teiid/query/processor/relational/TestProjectIntoNode.java 2012-06-27 19:35:38 UTC (rev 4209) @@ -39,6 +39,7 @@ import org.teiid.query.eval.Evaluator; import org.teiid.query.processor.FakeTupleSource; import org.teiid.query.processor.ProcessorDataManager; +import org.teiid.query.processor.RegisterRequestParameter; import org.teiid.query.processor.relational.ProjectIntoNode.Mode; import org.teiid.query.sql.lang.BatchedUpdateCommand; import org.teiid.query.sql.lang.Command; @@ -127,7 +128,7 @@ this.expectedBatchSize = expectedBatchSize; } public Object lookupCodeValue(CommandContext context,String codeTableName,String returnElementName,String keyElementName,Object keyValue) throws BlockedException,TeiidComponentException {return null;} - public TupleSource registerRequest(CommandContext context,Command command,String modelName,String connectorBindingId, int nodeID, int limit) throws TeiidComponentException, TeiidProcessingException { + public TupleSource registerRequest(CommandContext context,Command command,String modelName,RegisterRequestParameter parameterObject) throws TeiidComponentException, TeiidProcessingException { callCount++; int batchSize = 1;