[teiid-commits] teiid SVN: r2391 - in trunk: api/src/main/java/org/teiid/language/visitor and 27 other directories.
teiid-commits at lists.jboss.org
teiid-commits at lists.jboss.org
Thu Jul 29 11:18:37 EDT 2010
Author: shawkins
Date: 2010-07-29 11:18:33 -0400 (Thu, 29 Jul 2010)
New Revision: 2391
Added:
trunk/api/src/main/java/org/teiid/language/IteratorValueSource.java
trunk/engine/src/main/java/org/teiid/query/optimizer/BatchedUpdatePlanner.java
Removed:
trunk/engine/src/main/java/org/teiid/query/optimizer/batch/
trunk/engine/src/test/java/org/teiid/cache/
Modified:
trunk/api/src/main/java/org/teiid/language/visitor/AbstractLanguageVisitor.java
trunk/api/src/main/java/org/teiid/language/visitor/CollectorVisitor.java
trunk/api/src/main/java/org/teiid/language/visitor/LanguageObjectVisitor.java
trunk/api/src/main/java/org/teiid/language/visitor/SQLStringVisitor.java
trunk/api/src/main/java/org/teiid/translator/ExecutionFactory.java
trunk/build/kits/jboss-container/teiid-releasenotes.html
trunk/connectors/translator-jdbc/src/main/java/org/teiid/translator/jdbc/JDBCExecutionFactory.java
trunk/connectors/translator-jdbc/src/main/java/org/teiid/translator/jdbc/JDBCUpdateExecution.java
trunk/connectors/translator-jdbc/src/main/java/org/teiid/translator/jdbc/TranslatedCommand.java
trunk/connectors/translator-jdbc/src/main/java/org/teiid/translator/jdbc/modeshape/ModeShapeExecutionFactory.java
trunk/connectors/translator-jdbc/src/test/java/org/teiid/translator/jdbc/TestJDBCUpdateExecution.java
trunk/documentation/developer-guide/src/main/docbook/en-US/content/translator-api.xml
trunk/documentation/reference/src/main/docbook/en-US/content/translators.xml
trunk/engine/src/main/java/org/teiid/common/buffer/AbstractTupleSource.java
trunk/engine/src/main/java/org/teiid/common/buffer/STree.java
trunk/engine/src/main/java/org/teiid/common/buffer/TupleBrowser.java
trunk/engine/src/main/java/org/teiid/common/buffer/TupleBuffer.java
trunk/engine/src/main/java/org/teiid/common/buffer/TupleSource.java
trunk/engine/src/main/java/org/teiid/dqp/internal/datamgr/CapabilitiesConverter.java
trunk/engine/src/main/java/org/teiid/dqp/internal/datamgr/LanguageBridgeFactory.java
trunk/engine/src/main/java/org/teiid/dqp/internal/process/DataTierManagerImpl.java
trunk/engine/src/main/java/org/teiid/dqp/internal/process/DataTierTupleSource.java
trunk/engine/src/main/java/org/teiid/dqp/internal/process/PreparedStatementRequest.java
trunk/engine/src/main/java/org/teiid/dqp/internal/process/SessionAwareCache.java
trunk/engine/src/main/java/org/teiid/query/function/metadata/FunctionMethod.java
trunk/engine/src/main/java/org/teiid/query/function/source/SystemSource.java
trunk/engine/src/main/java/org/teiid/query/metadata/TempCapabilitiesFinder.java
trunk/engine/src/main/java/org/teiid/query/optimizer/QueryOptimizer.java
trunk/engine/src/main/java/org/teiid/query/optimizer/capabilities/SourceCapabilities.java
trunk/engine/src/main/java/org/teiid/query/optimizer/relational/PlanToProcessConverter.java
trunk/engine/src/main/java/org/teiid/query/processor/BatchIterator.java
trunk/engine/src/main/java/org/teiid/query/processor/CollectionTupleSource.java
trunk/engine/src/main/java/org/teiid/query/processor/proc/ProcedurePlan.java
trunk/engine/src/main/java/org/teiid/query/processor/relational/DependentCriteriaProcessor.java
trunk/engine/src/main/java/org/teiid/query/processor/relational/DependentValueSource.java
trunk/engine/src/main/java/org/teiid/query/processor/relational/GroupingNode.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/relational/SortNode.java
trunk/engine/src/main/java/org/teiid/query/processor/relational/SortUtility.java
trunk/engine/src/main/java/org/teiid/query/processor/relational/SortingFilter.java
trunk/engine/src/main/java/org/teiid/query/processor/relational/SourceState.java
trunk/engine/src/main/java/org/teiid/query/sql/lang/Insert.java
trunk/engine/src/main/java/org/teiid/query/sql/lang/Query.java
trunk/engine/src/main/java/org/teiid/query/sql/visitor/SQLStringVisitor.java
trunk/engine/src/main/java/org/teiid/query/tempdata/TempTable.java
trunk/engine/src/main/java/org/teiid/query/tempdata/TempTableStore.java
trunk/engine/src/test/java/org/teiid/dqp/internal/process/TestDQPCore.java
trunk/engine/src/test/java/org/teiid/query/optimizer/TestOptimizer.java
trunk/engine/src/test/java/org/teiid/query/processor/FakeTupleSource.java
trunk/engine/src/test/java/org/teiid/query/processor/TestCollectionTupleSource.java
trunk/engine/src/test/java/org/teiid/query/processor/relational/TestBatchedUpdateNode.java
trunk/engine/src/test/java/org/teiid/query/processor/relational/TestGroupingNode.java
trunk/engine/src/test/java/org/teiid/query/processor/relational/TestProjectIntoNode.java
trunk/engine/src/test/java/org/teiid/query/processor/relational/TestSortNode.java
Log:
TEIID-892 introduced a way better control large inserts. cleaned up the tuplesource interface. TEIID-1167 changed temp tables to support the iterator based inserts since they do not support bulk delete/update logic.
Added: trunk/api/src/main/java/org/teiid/language/IteratorValueSource.java
===================================================================
--- trunk/api/src/main/java/org/teiid/language/IteratorValueSource.java (rev 0)
+++ trunk/api/src/main/java/org/teiid/language/IteratorValueSource.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -0,0 +1,58 @@
+/*
+ * 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.language;
+
+import java.util.Iterator;
+import java.util.List;
+
+import org.teiid.language.visitor.LanguageObjectVisitor;
+
+public class IteratorValueSource<T extends List<?>> extends BaseLanguageObject implements InsertValueSource {
+
+ private Iterator<T> iter;
+ private int columnCount;
+
+ public IteratorValueSource(Iterator<T> iter, int columnCount) {
+ this.iter = iter;
+ this.columnCount = columnCount;
+ }
+
+ /**
+ * A memory safe iterator of the insert values. Only 1 iterator is associated
+ * with the value source. Once it is consumed there are no more values.
+ * @return
+ */
+ public Iterator<T> getIterator() {
+ return iter;
+ }
+
+ public int getColumnCount() {
+ return columnCount;
+ }
+
+ @Override
+ public void acceptVisitor(LanguageObjectVisitor visitor) {
+ visitor.visit(this);
+ }
+
+}
Property changes on: trunk/api/src/main/java/org/teiid/language/IteratorValueSource.java
___________________________________________________________________
Name: svn:mime-type
+ text/plain
Modified: trunk/api/src/main/java/org/teiid/language/visitor/AbstractLanguageVisitor.java
===================================================================
--- trunk/api/src/main/java/org/teiid/language/visitor/AbstractLanguageVisitor.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/api/src/main/java/org/teiid/language/visitor/AbstractLanguageVisitor.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -41,6 +41,7 @@
import org.teiid.language.In;
import org.teiid.language.Insert;
import org.teiid.language.IsNull;
+import org.teiid.language.IteratorValueSource;
import org.teiid.language.Join;
import org.teiid.language.LanguageObject;
import org.teiid.language.Like;
@@ -146,4 +147,5 @@
public void visit(SetQuery obj) {}
public void visit(SetClause obj) {}
public void visit(SearchedWhenClause obj) {}
+ public void visit(IteratorValueSource obj) {}
}
Modified: trunk/api/src/main/java/org/teiid/language/visitor/CollectorVisitor.java
===================================================================
--- trunk/api/src/main/java/org/teiid/language/visitor/CollectorVisitor.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/api/src/main/java/org/teiid/language/visitor/CollectorVisitor.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -44,6 +44,7 @@
import org.teiid.language.In;
import org.teiid.language.Insert;
import org.teiid.language.IsNull;
+import org.teiid.language.IteratorValueSource;
import org.teiid.language.Join;
import org.teiid.language.LanguageObject;
import org.teiid.language.Like;
@@ -226,6 +227,11 @@
public void visit(SearchedWhenClause obj) {
checkInstance(obj);
}
+
+ @Override
+ public void visit(IteratorValueSource obj) {
+ checkInstance(obj);
+ }
/**
* This is a utility method to instantiate and run the visitor in conjunction
Modified: trunk/api/src/main/java/org/teiid/language/visitor/LanguageObjectVisitor.java
===================================================================
--- trunk/api/src/main/java/org/teiid/language/visitor/LanguageObjectVisitor.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/api/src/main/java/org/teiid/language/visitor/LanguageObjectVisitor.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -61,4 +61,5 @@
public void visit(SetQuery obj);
public void visit(SetClause obj);
public void visit(SearchedWhenClause obj);
+ public void visit(IteratorValueSource obj);
}
Modified: trunk/api/src/main/java/org/teiid/language/visitor/SQLStringVisitor.java
===================================================================
--- trunk/api/src/main/java/org/teiid/language/visitor/SQLStringVisitor.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/api/src/main/java/org/teiid/language/visitor/SQLStringVisitor.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -50,6 +50,7 @@
import org.teiid.language.In;
import org.teiid.language.Insert;
import org.teiid.language.IsNull;
+import org.teiid.language.IteratorValueSource;
import org.teiid.language.Join;
import org.teiid.language.LanguageObject;
import org.teiid.language.Like;
@@ -494,6 +495,18 @@
append(obj.getValues());
buffer.append(Tokens.RPAREN);
}
+
+ @Override
+ public void visit(IteratorValueSource obj) {
+ buffer.append(VALUES).append(Tokens.SPACE).append(Tokens.LPAREN);
+ for (int i = 0; i < obj.getColumnCount(); i++) {
+ if (i > 0) {
+ buffer.append(", "); //$NON-NLS-1$
+ }
+ buffer.append("?"); //$NON-NLS-1$
+ }
+ buffer.append(Tokens.RPAREN);
+ }
public void visit(IsNull obj) {
append(obj.getExpression());
Modified: trunk/api/src/main/java/org/teiid/translator/ExecutionFactory.java
===================================================================
--- trunk/api/src/main/java/org/teiid/translator/ExecutionFactory.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/api/src/main/java/org/teiid/translator/ExecutionFactory.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -23,6 +23,7 @@
package org.teiid.translator;
import java.util.Collection;
+import java.util.Iterator;
import java.util.List;
import javax.resource.ResourceException;
@@ -36,6 +37,7 @@
import org.teiid.language.Command;
import org.teiid.language.LanguageFactory;
import org.teiid.language.QueryExpression;
+import org.teiid.language.Select;
import org.teiid.language.SetQuery;
import org.teiid.logging.LogConstants;
import org.teiid.logging.LogManager;
@@ -730,12 +732,22 @@
/**
* Support indicates that the connector can accept INSERTs with
- * values specified by an {@link SetQuery}.
+ * values specified by a {@link SetQuery} or {@link Select}
* @since 6.1
*/
public boolean supportsInsertWithQueryExpression() {
return false;
}
+
+ /**
+ * Support indicates that the connector can accept INSERTs
+ * with values specified by an {@link Iterator}
+ * @since 7.1
+ * @return
+ */
+ public boolean supportsInsertWithIterator() {
+ return false;
+ }
public static <T> T getInstance(Class<T> expectedType, String className, Collection<?> ctorObjs, Class<? extends T> defaultClass) throws TranslatorException {
try {
Modified: trunk/build/kits/jboss-container/teiid-releasenotes.html
===================================================================
--- trunk/build/kits/jboss-container/teiid-releasenotes.html 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/build/kits/jboss-container/teiid-releasenotes.html 2010-07-29 15:18:33 UTC (rev 2391)
@@ -10,7 +10,7 @@
<P><A HREF="http://www.teiid.org/"><IMG SRC="https://www.jboss.org/dms/teiid/images/teiid_banner.png" NAME="graphics1" ALT="Teiid" ALIGN=BOTTOM WIDTH=800></A>
<H1>Teiid ${project.version} Release Notes</H1>
-<P>Teiid ${project.version} adds ODBC, SQL, and caching features.
+<P>Teiid ${project.version} adds ODBC, SQL, performance, and caching features.
<H2>Overview</H2>
<UL>
@@ -38,7 +38,11 @@
<li>Added PRIMARY KEY and the associated index support to temp tables.
</ul>
<LI><B>Parallel Source Queries</B> - reestablished parallel execution of source queries within a query plan along with a prioritized work system to help prevent resource contention.
+ <LI><B>Improved Insert Support</B> - Cross source inserts using a query expression can defer to the source to perform the entire insert atomically with source controlled batching.
+ See maxPreparedInsertBatchSize on any JDBC translator supporting the InsertWithIterator capability.
<LI><B>SHOW Statement</B> - added client handling for the SHOW statement to retrieve query plan information and see parameter values.
+ <LI><B>User Identity</B> - Teiid DataSources in JBossAS connecting to a local Teiid instance can preserve the thread's user identity, rather than requiring
+ a new authentication. See TeiidDataSource.setPassthroughAuthentication.
</UL>
<h2><a name="Compatibility">Compatibility Issues</a></h2>
Modified: trunk/connectors/translator-jdbc/src/main/java/org/teiid/translator/jdbc/JDBCExecutionFactory.java
===================================================================
--- trunk/connectors/translator-jdbc/src/main/java/org/teiid/translator/jdbc/JDBCExecutionFactory.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/connectors/translator-jdbc/src/main/java/org/teiid/translator/jdbc/JDBCExecutionFactory.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -139,6 +139,7 @@
private boolean trimStrings;
private boolean useCommentsInSourceQuery;
private String version;
+ private int maxInsertBatchSize = 2048;
private AtomicBoolean initialConnection = new AtomicBoolean(true);
@@ -440,6 +441,11 @@
}
@Override
+ public boolean supportsInsertWithIterator() {
+ return super.supportsBulkUpdate();
+ }
+
+ @Override
public boolean supportsBatchedUpdates() {
return true;
}
@@ -462,9 +468,25 @@
@Override
public boolean supportsInsertWithQueryExpression() {
return true;
- }
+ }
/**
+ * Get the max number of inserts to perform in one batch.
+ * @return
+ */
+ @TranslatorProperty(display="Max Prepared Insert Batch Size", description="The max size of a prepared insert batch. Default 2048.", advanced=true)
+ public int getMaxPreparedInsertBatchSize() {
+ return maxInsertBatchSize;
+ }
+
+ public void setMaxPreparedInsertBatchSize(int maxInsertBatchSize) {
+ if (maxInsertBatchSize < 1) {
+ throw new AssertionError("Max prepared batch insert size must be greater than 0"); //$NON-NLS-1$
+ }
+ this.maxInsertBatchSize = maxInsertBatchSize;
+ }
+
+ /**
* Gets the database calendar. This will be set to the time zone
* specified by the property {@link JDBCPropertyNames#DATABASE_TIME_ZONE}, or
* the local time zone if none is specified.
Modified: trunk/connectors/translator-jdbc/src/main/java/org/teiid/translator/jdbc/JDBCUpdateExecution.java
===================================================================
--- trunk/connectors/translator-jdbc/src/main/java/org/teiid/translator/jdbc/JDBCUpdateExecution.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/connectors/translator-jdbc/src/main/java/org/teiid/translator/jdbc/JDBCUpdateExecution.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -26,14 +26,17 @@
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
+import java.util.Iterator;
import java.util.List;
import org.teiid.language.BatchedUpdates;
import org.teiid.language.Command;
+import org.teiid.language.Insert;
+import org.teiid.language.IteratorValueSource;
import org.teiid.language.Literal;
-import org.teiid.translator.TranslatorException;
import org.teiid.translator.DataNotAvailableException;
import org.teiid.translator.ExecutionContext;
+import org.teiid.translator.TranslatorException;
import org.teiid.translator.UpdateExecution;
@@ -163,8 +166,46 @@
int updateCount = 0;
if (!translatedComm.isPrepared()) {
updateCount = getStatement().executeUpdate(sql);
+ addStatementWarnings();
} else {
PreparedStatement pstatement = getPreparedStatement(sql);
+
+ if (command instanceof Insert) {
+ Insert insert = (Insert)command;
+ if (insert.getValueSource() instanceof IteratorValueSource) {
+ commitType = getAutoCommit(translatedComm);
+ if (commitType) {
+ connection.setAutoCommit(false);
+ }
+
+ IteratorValueSource<List<Object>> ivs = (IteratorValueSource)insert.getValueSource();
+ List<Object>[] values = new List[ivs.getColumnCount()];
+ for (int i = 0; i < ivs.getColumnCount(); i++) {
+ values[i] = new ArrayList<Object>();
+ Literal literal = new Literal(values[i], insert.getColumns().get(i).getType());
+ literal.setMultiValued(true);
+ translatedComm.getPreparedValues().add(literal);
+ }
+ Iterator<List<Object>> i = ivs.getIterator();
+ int maxBatchSize = this.executionFactory.getMaxPreparedInsertBatchSize();
+ while (i.hasNext()) {
+ int batchSize = 0;
+ while (i.hasNext() && batchSize++ < maxBatchSize) {
+ List<Object> next = i.next();
+ for (int j = 0; j < ivs.getColumnCount(); j++) {
+ values[j].add(next.get(j));
+ }
+ }
+ updateCount += executePreparedBatch(translatedComm, pstatement, batchSize);
+ for (int j = 0; j < ivs.getColumnCount(); j++) {
+ values[j].clear();
+ }
+ }
+ succeeded = true;
+ return new int[updateCount];
+ }
+ }
+
int rowCount = 1;
for (int i = 0; i< translatedComm.getPreparedValues().size(); i++) {
Literal paramValue = (Literal)translatedComm.getPreparedValues().get(i);
@@ -179,19 +220,9 @@
connection.setAutoCommit(false);
}
}
- bindPreparedStatementValues(pstatement, translatedComm, rowCount);
- if (rowCount > 1) {
- int[] results = pstatement.executeBatch();
-
- for (int i=0; i<results.length; i++) {
- updateCount += results[i];
- }
- succeeded = true;
- } else {
- updateCount = pstatement.executeUpdate();
- }
+ updateCount = executePreparedBatch(translatedComm, pstatement, rowCount);
+ succeeded = true;
}
- addStatementWarnings();
return new int[] {updateCount};
} catch (SQLException err) {
throw new JDBCExecutionException(err, translatedComm);
@@ -202,6 +233,23 @@
}
}
+ private int executePreparedBatch(TranslatedCommand translatedComm, PreparedStatement pstatement, int rowCount)
+ throws SQLException {
+ bindPreparedStatementValues(pstatement, translatedComm, rowCount);
+ int updateCount = 0;
+ if (rowCount > 1) {
+ int[] results = pstatement.executeBatch();
+
+ for (int i=0; i<results.length; i++) {
+ updateCount += results[i];
+ }
+ } else {
+ updateCount = pstatement.executeUpdate();
+ }
+ addStatementWarnings();
+ return updateCount;
+ }
+
/**
* @param command
* @return
Modified: trunk/connectors/translator-jdbc/src/main/java/org/teiid/translator/jdbc/TranslatedCommand.java
===================================================================
--- trunk/connectors/translator-jdbc/src/main/java/org/teiid/translator/jdbc/TranslatedCommand.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/connectors/translator-jdbc/src/main/java/org/teiid/translator/jdbc/TranslatedCommand.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -27,6 +27,8 @@
import java.util.List;
import org.teiid.language.Command;
+import org.teiid.language.Insert;
+import org.teiid.language.IteratorValueSource;
import org.teiid.language.Literal;
import org.teiid.language.visitor.CollectorVisitor;
import org.teiid.translator.TranslatorException;
@@ -89,6 +91,10 @@
return true;
}
}
+ if (command instanceof Insert) {
+ Insert insert = (Insert)command;
+ return insert.getValueSource() instanceof IteratorValueSource<?>;
+ }
return false;
}
Modified: trunk/connectors/translator-jdbc/src/main/java/org/teiid/translator/jdbc/modeshape/ModeShapeExecutionFactory.java
===================================================================
--- trunk/connectors/translator-jdbc/src/main/java/org/teiid/translator/jdbc/modeshape/ModeShapeExecutionFactory.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/connectors/translator-jdbc/src/main/java/org/teiid/translator/jdbc/modeshape/ModeShapeExecutionFactory.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -30,7 +30,6 @@
import java.util.List;
import org.teiid.language.ColumnReference;
-import org.teiid.language.Command;
import org.teiid.language.Expression;
import org.teiid.language.Function;
import org.teiid.language.LanguageObject;
@@ -39,7 +38,6 @@
import org.teiid.translator.ExecutionContext;
import org.teiid.translator.Translator;
import org.teiid.translator.TranslatorException;
-import org.teiid.translator.TranslatorProperty;
import org.teiid.translator.jdbc.ConvertModifier;
import org.teiid.translator.jdbc.FunctionModifier;
import org.teiid.translator.jdbc.JDBCExecutionFactory;
@@ -52,8 +50,10 @@
@Translator(name="modeshape")
public class ModeShapeExecutionFactory extends JDBCExecutionFactory {
- private String version = "2.0"; //$NON-NLS-1$
-
+ public ModeShapeExecutionFactory() {
+ setDatabaseVersion("2.0"); //$NON-NLS-1$
+ }
+
@Override
public void start() throws TranslatorException {
super.start();
@@ -160,16 +160,7 @@
return "FALSE"; //$NON-NLS-1$
}
-
-
@Override
- public List<?> translateCommand(Command command, ExecutionContext context) {
- return super.translateCommand(command, context);
- }
-
-
-
- @Override
public String translateLiteralDate(Date dateValue) {
return "DATE '" + formatDateValue(dateValue) + "'"; //$NON-NLS-1$//$NON-NLS-2$
}
@@ -201,15 +192,6 @@
}
- @TranslatorProperty(description= "ModeShape Repository Version")
- public String getDatabaseVersion() {
- return this.version;
- }
-
- public void setDatabaseVersion(String version) {
- this.version = version;
- }
-
@Override
public boolean useBindVariables() {
return false;
Modified: trunk/connectors/translator-jdbc/src/test/java/org/teiid/translator/jdbc/TestJDBCUpdateExecution.java
===================================================================
--- trunk/connectors/translator-jdbc/src/test/java/org/teiid/translator/jdbc/TestJDBCUpdateExecution.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/connectors/translator-jdbc/src/test/java/org/teiid/translator/jdbc/TestJDBCUpdateExecution.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -24,13 +24,16 @@
import java.sql.Connection;
import java.sql.PreparedStatement;
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.List;
import org.junit.Test;
import org.mockito.Mockito;
import org.teiid.language.Command;
import org.teiid.language.ExpressionValueSource;
import org.teiid.language.Insert;
+import org.teiid.language.IteratorValueSource;
import org.teiid.language.Literal;
import org.teiid.translator.ExecutionContext;
@@ -58,4 +61,23 @@
Mockito.verify(p, Mockito.times(2)).addBatch();
}
+ @Test public void testInsertIteratorUpdate() throws Exception {
+ Insert command = (Insert)TranslationHelper.helpTranslate(TranslationHelper.BQT_VDB, "insert into BQT1.SmallA (IntKey, IntNum) values (1, 2)"); //$NON-NLS-1$
+ List<List<Integer>> values = new ArrayList<List<Integer>>();
+ values.add(Arrays.asList(1, 2));
+ values.add(Arrays.asList(2, 3));
+ command.setValueSource(new IteratorValueSource(values.iterator(), 2));
+
+ Connection connection = Mockito.mock(Connection.class);
+ PreparedStatement p = Mockito.mock(PreparedStatement.class);
+ Mockito.stub(p.executeBatch()).toReturn(new int [] {1, 1});
+ Mockito.stub(connection.prepareStatement("INSERT INTO SmallA (IntKey, IntNum) VALUES (?, ?)")).toReturn(p); //$NON-NLS-1$
+
+ JDBCExecutionFactory config = new JDBCExecutionFactory();
+
+ JDBCUpdateExecution updateExecution = new JDBCUpdateExecution(command, connection, Mockito.mock(ExecutionContext.class), config);
+ updateExecution.execute();
+ Mockito.verify(p, Mockito.times(2)).addBatch();
+ }
+
}
Modified: trunk/documentation/developer-guide/src/main/docbook/en-US/content/translator-api.xml
===================================================================
--- trunk/documentation/developer-guide/src/main/docbook/en-US/content/translator-api.xml 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/documentation/developer-guide/src/main/docbook/en-US/content/translator-api.xml 2010-07-29 15:18:33 UTC (rev 2391)
@@ -466,8 +466,8 @@
<para>Each <code>Insert</code> will have a single <code>NamedTable</code> specifying the table being
inserted into. It will also has a list of <code>ColumnReference</code> specifying the columns
of the <code>NamedTable</code> that are being inserted into. It also has
- <code>InsertValueSource</code>, which will either be a list of
- <code>Expression(ExpressionValueSource)</code> or <code>QueryExpression</code></para>
+ <code>InsertValueSource</code>, which will be a list of
+ Expressions (<code>ExpressionValueSource</code>), or a <code>QueryExpression</code>, or an Iterator (<code>IteratorValueSource</code>)</para>
</sect3>
<sect3>
@@ -1245,7 +1245,7 @@
</row>
<row>
<entry>
- <para>supportsBulkUpdate</para>
+ <para>BulkUpdate</para>
</entry>
<entry>
<para/>
@@ -1254,6 +1254,17 @@
<para>Translator supports updates with multiple value sets</para>
</entry>
</row>
+ <row>
+ <entry>
+ <para>InsertWithIterator</para>
+ </entry>
+ <entry>
+ <para/>
+ </entry>
+ <entry>
+ <para>Translator supports inserts with an iterator of values. The values would typically be from an evaluated QueryExpression.</para>
+ </entry>
+ </row>
</tbody>
</tgroup>
</table>
Modified: trunk/documentation/reference/src/main/docbook/en-US/content/translators.xml
===================================================================
--- trunk/documentation/reference/src/main/docbook/en-US/content/translators.xml 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/documentation/reference/src/main/docbook/en-US/content/translators.xml 2010-07-29 15:18:33 UTC (rev 2391)
@@ -196,6 +196,11 @@
<entry>This will embed a /*comment*/ leading comment with session/request id in source SQL query for informational purposes</entry>
<entry>false</entry>
</row>
+<row>
+<entry>MaxPreparedInsertBatchSize</entry>
+<entry>The max size of a prepared insert batch.</entry>
+<entry>2048</entry>
+</row>
</tbody>
</tgroup>
</table>
Modified: trunk/engine/src/main/java/org/teiid/common/buffer/AbstractTupleSource.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/common/buffer/AbstractTupleSource.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/common/buffer/AbstractTupleSource.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -73,6 +73,8 @@
protected abstract TupleBatch getBatch(int row) throws TeiidComponentException, TeiidProcessingException;
+ protected abstract int available();
+
@Override
public void closeSource() {
batch = null;
Modified: trunk/engine/src/main/java/org/teiid/common/buffer/STree.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/common/buffer/STree.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/common/buffer/STree.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -35,10 +35,8 @@
/**
* Self balancing search tree using skip list like logic
- * This has similar performance similar to a B+/-Tree
- *
- * TODO: reserve additional memory for delete/update operations
- * TODO: double link to support desc key access
+ * This has similar performance similar to a B+/-Tree,
+ * but with fewer updates.
*/
@SuppressWarnings("unchecked")
public class STree {
@@ -67,7 +65,6 @@
int keyLength,
String[] types) {
randomSeed = seedGenerator.nextInt() | 0x00000100; // ensure nonzero
- randomSeed = 1;
this.keyManager = manager;
this.leafManager = leafManager;
this.comparator = comparator;
@@ -327,6 +324,10 @@
return oldSize;
}
+ public int getHeight() {
+ return header.length;
+ }
+
@Override
public String toString() {
StringBuffer result = new StringBuffer();
Modified: trunk/engine/src/main/java/org/teiid/common/buffer/TupleBrowser.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/common/buffer/TupleBrowser.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/common/buffer/TupleBrowser.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -28,11 +28,12 @@
import org.teiid.common.buffer.SPage.SearchResult;
import org.teiid.core.TeiidComponentException;
+import org.teiid.core.TeiidProcessingException;
/**
* Implements intelligent browsing over a {@link STree}
*/
-public class TupleBrowser {
+public class TupleBrowser implements TupleSource {
private final STree tree;
@@ -132,12 +133,9 @@
return result;
}
- /**
- * Returns the next tuple or null if there are no more results.
- * @return
- * @throws TeiidComponentException
- */
- public List<?> next() throws TeiidComponentException {
+ @Override
+ public List<?> nextTuple() throws TeiidComponentException,
+ TeiidProcessingException {
for (;;) {
//first check for value iteration
if (valueSet != null) {
@@ -237,4 +235,10 @@
public void removed() {
index-=getOffset();
}
+
+ @Override
+ public void closeSource() {
+
+ }
+
}
\ No newline at end of file
Modified: trunk/engine/src/main/java/org/teiid/common/buffer/TupleBuffer.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/common/buffer/TupleBuffer.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/common/buffer/TupleBuffer.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -268,14 +268,12 @@
* @return
*/
public IndexedTupleSource createIndexedTupleSource(final boolean singleUse) {
+ if (singleUse) {
+ setForwardOnly(true);
+ }
return new AbstractTupleSource() {
@Override
- public List<? extends Expression> getSchema() {
- return (List<? extends Expression>) schema;
- }
-
- @Override
protected List<?> finalRow() throws BlockedException {
if(isFinal) {
return null;
Modified: trunk/engine/src/main/java/org/teiid/common/buffer/TupleSource.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/common/buffer/TupleSource.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/common/buffer/TupleSource.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -26,7 +26,6 @@
import org.teiid.core.TeiidComponentException;
import org.teiid.core.TeiidProcessingException;
-import org.teiid.query.sql.symbol.Expression;
/**
@@ -37,12 +36,6 @@
public interface TupleSource {
/**
- * Returns the List of ElementSymbol describing the Tuple Source
- * @return the List of elements describing the Tuple Source
- */
- List<? extends Expression> getSchema();
-
- /**
* Returns the next tuple
* @return the next tuple (a List object), or <code>null</code> if
* there are no more tuples.
@@ -58,10 +51,4 @@
*/
void closeSource();
- /**
- * Returns an estimate of the number of rows that can be read without blocking.
- * @return
- */
- int available();
-
}
Modified: trunk/engine/src/main/java/org/teiid/dqp/internal/datamgr/CapabilitiesConverter.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/dqp/internal/datamgr/CapabilitiesConverter.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/dqp/internal/datamgr/CapabilitiesConverter.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -96,7 +96,7 @@
tgtCaps.setCapabilitySupport(Capability.QUERY_ORDERBY_UNRELATED, srcCaps.supportsOrderByUnrelated());
tgtCaps.setCapabilitySupport(Capability.QUERY_AGGREGATES_ENHANCED_NUMERIC, srcCaps.supportsAggregatesEnhancedNumeric());
tgtCaps.setCapabilitySupport(Capability.QUERY_ORDERBY_NULL_ORDERING, srcCaps.supportsOrderByNullOrdering());
-
+ tgtCaps.setCapabilitySupport(Capability.INSERT_WITH_ITERATOR, srcCaps.supportsInsertWithIterator());
List functions = srcCaps.getSupportedFunctions();
if(functions != null && functions.size() > 0) {
Iterator iter = functions.iterator();
Modified: trunk/engine/src/main/java/org/teiid/dqp/internal/datamgr/LanguageBridgeFactory.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/dqp/internal/datamgr/LanguageBridgeFactory.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/dqp/internal/datamgr/LanguageBridgeFactory.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -25,10 +25,13 @@
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
+import java.util.NoSuchElementException;
import org.teiid.api.exception.query.QueryMetadataException;
import org.teiid.client.metadata.ParameterInfo;
+import org.teiid.common.buffer.TupleSource;
import org.teiid.core.TeiidComponentException;
+import org.teiid.core.TeiidException;
import org.teiid.core.TeiidRuntimeException;
import org.teiid.language.AggregateFunction;
import org.teiid.language.AndOr;
@@ -44,6 +47,7 @@
import org.teiid.language.In;
import org.teiid.language.InsertValueSource;
import org.teiid.language.IsNull;
+import org.teiid.language.IteratorValueSource;
import org.teiid.language.Join;
import org.teiid.language.Like;
import org.teiid.language.Literal;
@@ -540,6 +544,40 @@
InsertValueSource valueSource = null;
if (insert.getQueryExpression() != null) {
valueSource = translate(insert.getQueryExpression());
+ } else if (insert.getTupleSource() != null) {
+ final TupleSource ts = insert.getTupleSource();
+ valueSource = new IteratorValueSource<List<?>>(new Iterator<List<?>>() {
+
+ List<?> next;
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public List<?> next() {
+ if (hasNext()) {
+ List<?> result = next;
+ next = null;
+ return result;
+ }
+ throw new NoSuchElementException();
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (next != null) {
+ return true;
+ }
+ try {
+ next = ts.nextTuple();
+ } catch (TeiidException e) {
+ throw new TeiidRuntimeException(e);
+ }
+ return next != null;
+ }
+ }, elements.size());
} else {
// This is for the simple one row insert.
List values = insert.getValues();
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 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/dqp/internal/process/DataTierManagerImpl.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -307,7 +307,7 @@
break;
}
}
- return new CollectionTupleSource(rows.iterator(), command.getProjectedSymbols());
+ return new CollectionTupleSource(rows.iterator());
}
private List<Schema> getVisibleSchemas(VDBMetaData vdb, CompositeMetadataStore metadata) {
Modified: trunk/engine/src/main/java/org/teiid/dqp/internal/process/DataTierTupleSource.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/dqp/internal/process/DataTierTupleSource.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/dqp/internal/process/DataTierTupleSource.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -299,12 +299,4 @@
return this.aqr.isTransactional();
}
- @Override
- public int available() {
- if (this.arm == null) {
- return 0;
- }
- return this.arm.getResults().length - index;
- }
-
}
Modified: trunk/engine/src/main/java/org/teiid/dqp/internal/process/PreparedStatementRequest.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/dqp/internal/process/PreparedStatementRequest.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/dqp/internal/process/PreparedStatementRequest.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -41,7 +41,7 @@
import org.teiid.query.QueryPlugin;
import org.teiid.query.eval.Evaluator;
import org.teiid.query.metadata.QueryMetadataInterface;
-import org.teiid.query.optimizer.batch.BatchedUpdatePlanner;
+import org.teiid.query.optimizer.BatchedUpdatePlanner;
import org.teiid.query.optimizer.capabilities.SourceCapabilities;
import org.teiid.query.processor.ProcessorPlan;
import org.teiid.query.processor.relational.AccessNode;
Modified: trunk/engine/src/main/java/org/teiid/dqp/internal/process/SessionAwareCache.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/dqp/internal/process/SessionAwareCache.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/dqp/internal/process/SessionAwareCache.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -28,7 +28,6 @@
import java.util.concurrent.atomic.AtomicInteger;
import org.teiid.cache.Cache;
-import org.teiid.cache.CacheConfiguration;
import org.teiid.cache.CacheFactory;
import org.teiid.cache.DefaultCache;
import org.teiid.cache.DefaultCacheFactory;
Modified: trunk/engine/src/main/java/org/teiid/query/function/metadata/FunctionMethod.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/query/function/metadata/FunctionMethod.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/query/function/metadata/FunctionMethod.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -69,7 +69,7 @@
/*
* always -> normal deterministic functions
- * server lifetime -> lookup (however lookup values can be flushed at any time)
+ * vdb -> lookup (however lookup values can be flushed at any time), current_database
* session -> env, user
* command -> command payload
* never -> rand, etc.
@@ -83,7 +83,7 @@
* processing time.
*/
public static final int DETERMINISTIC = 0;
- public static final int SERVER_DETERMINISTIC = 1;
+ public static final int VDB_DETERMINISTIC = 1;
public static final int USER_DETERMINISTIC = 2;
public static final int SESSION_DETERMINISTIC = 3;
public static final int COMMAND_DETERMINISTIC = 4;
Modified: trunk/engine/src/main/java/org/teiid/query/function/source/SystemSource.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/query/function/source/SystemSource.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/query/function/source/SystemSource.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -762,7 +762,7 @@
new FunctionParameter("keyelement", DataTypeManager.DefaultDataTypes.STRING, QueryPlugin.Util.getString("SystemSource.Lookup_arg3")), //$NON-NLS-1$ //$NON-NLS-2$
new FunctionParameter("keyvalue", keyValueType, QueryPlugin.Util.getString("SystemSource.Lookup_arg4")), //$NON-NLS-1$ //$NON-NLS-2$
},
- new FunctionParameter("result", DataTypeManager.DefaultDataTypes.OBJECT, QueryPlugin.Util.getString("SystemSource.Lookup_result")), true, FunctionMethod.SERVER_DETERMINISTIC ) ); //$NON-NLS-1$ //$NON-NLS-2$
+ new FunctionParameter("result", DataTypeManager.DefaultDataTypes.OBJECT, QueryPlugin.Util.getString("SystemSource.Lookup_result")), true, FunctionMethod.VDB_DETERMINISTIC ) ); //$NON-NLS-1$ //$NON-NLS-2$
}
}
@@ -773,11 +773,9 @@
}
private void addCurrentDatabaseFunction() {
- // this function is actually session_deterministic, however for caching purposes it has been set deterministic
- // as caching is already scoped to the VDB.
functions.add(
new FunctionMethod("current_database", QueryPlugin.Util.getString("SystemSource.current_database_desc"), MISCELLANEOUS, FunctionMethod.CANNOT_PUSHDOWN, FUNCTION_CLASS, "current_database", null, //$NON-NLS-1$ //$NON-NLS-2$ //$NON-NLS-3$
- new FunctionParameter("result", DataTypeManager.DefaultDataTypes.STRING, QueryPlugin.Util.getString("current_database_result")), false, FunctionMethod.DETERMINISTIC) ); //$NON-NLS-1$ //$NON-NLS-2$
+ new FunctionParameter("result", DataTypeManager.DefaultDataTypes.STRING, QueryPlugin.Util.getString("current_database_result")), false, FunctionMethod.VDB_DETERMINISTIC) ); //$NON-NLS-1$ //$NON-NLS-2$
}
private void addEnvFunction() {
Modified: trunk/engine/src/main/java/org/teiid/query/metadata/TempCapabilitiesFinder.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/query/metadata/TempCapabilitiesFinder.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/query/metadata/TempCapabilitiesFinder.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -43,7 +43,7 @@
if (TempMetadataAdapter.TEMP_MODEL.getID().equals(modelName)) {
if (tempCaps == null) {
tempCaps = new BasicSourceCapabilities();
- tempCaps.setCapabilitySupport(Capability.BULK_UPDATE, true);
+ tempCaps.setCapabilitySupport(Capability.INSERT_WITH_ITERATOR, true);
tempCaps.setCapabilitySupport(Capability.QUERY_ORDERBY, true);
tempCaps.setCapabilitySupport(Capability.CRITERIA_NOT, true);
tempCaps.setCapabilitySupport(Capability.CRITERIA_IN, true);
Copied: trunk/engine/src/main/java/org/teiid/query/optimizer/BatchedUpdatePlanner.java (from rev 2386, trunk/engine/src/main/java/org/teiid/query/optimizer/batch/BatchedUpdatePlanner.java)
===================================================================
--- trunk/engine/src/main/java/org/teiid/query/optimizer/BatchedUpdatePlanner.java (rev 0)
+++ trunk/engine/src/main/java/org/teiid/query/optimizer/BatchedUpdatePlanner.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -0,0 +1,236 @@
+/*
+ * 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.optimizer;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.teiid.api.exception.query.QueryMetadataException;
+import org.teiid.api.exception.query.QueryPlannerException;
+import org.teiid.core.TeiidComponentException;
+import org.teiid.core.TeiidRuntimeException;
+import org.teiid.core.id.IDGenerator;
+import org.teiid.core.id.IntegerID;
+import org.teiid.query.analysis.AnalysisRecord;
+import org.teiid.query.execution.QueryExecPlugin;
+import org.teiid.query.metadata.QueryMetadataInterface;
+import org.teiid.query.optimizer.capabilities.CapabilitiesFinder;
+import org.teiid.query.optimizer.capabilities.SourceCapabilities;
+import org.teiid.query.optimizer.capabilities.SourceCapabilities.Capability;
+import org.teiid.query.optimizer.relational.rules.CapabilitiesUtil;
+import org.teiid.query.processor.ProcessorPlan;
+import org.teiid.query.processor.batch.BatchedUpdatePlan;
+import org.teiid.query.processor.relational.BatchedUpdateNode;
+import org.teiid.query.processor.relational.ProjectNode;
+import org.teiid.query.processor.relational.RelationalPlan;
+import org.teiid.query.sql.lang.BatchedUpdateCommand;
+import org.teiid.query.sql.lang.Command;
+import org.teiid.query.sql.lang.Delete;
+import org.teiid.query.sql.lang.Insert;
+import org.teiid.query.sql.lang.Update;
+import org.teiid.query.sql.symbol.GroupSymbol;
+import org.teiid.query.sql.util.VariableContext;
+import org.teiid.query.sql.visitor.EvaluatableVisitor;
+import org.teiid.query.util.CommandContext;
+
+
+
+/**
+ * Planner for BatchedUpdateCommands
+ * @since 4.2
+ */
+public class BatchedUpdatePlanner implements CommandPlanner {
+
+ /**
+ * Optimizes batched updates by batching all contiguous commands that relate to the same physical model.
+ * For example, for the following batch of commands:
+ * <br/>
+ * <ol>
+ * <li>1. INSERT INTO physicalModel.myPhysical ...</li>
+ * <li>2. UPDATE physicalModel.myPhysical ... </li>
+ * <li>3. DELETE FROM virtualmodel.myVirtual ... </li>
+ * <li>4. UPDATE virtualmodel.myVirtual ... </li>
+ * <li>5. UPDATE physicalModel.myOtherPhysical ...</li>
+ * <li>6. INSERT INTO physicalModel.myOtherPhysical ... <li>
+ * <li>7. DELETE FROM physicalModel.myOtherPhysical ...</li>
+ * <li>8. INSERT INTO physicalModel.myPhysical ... </li>
+ * <li>9. INSERT INTO physicalModel.myPhysical ... </li>
+ * <li>10. INSERT INTO physicalModel.myPhysical ... </li>
+ * <li>11. INSERT INTO physicalModel.myPhysical ... </li>
+ * <li>12. INSERT INTO physicalModel.myPhysical ... </li>
+ * </ol>
+ * <br/> this implementation will batch as follows: (1,2), (5, 6, 7), (8 thru 12).
+ * The remaining commands/plans will be executed individually.
+ * @see org.teiid.query.optimizer.CommandPlanner#optimize(Command, org.teiid.core.id.IDGenerator, org.teiid.query.metadata.QueryMetadataInterface, org.teiid.query.optimizer.capabilities.CapabilitiesFinder, org.teiid.query.analysis.AnalysisRecord, CommandContext)
+ * @since 4.2
+ */
+ public ProcessorPlan optimize(Command command,
+ IDGenerator idGenerator,
+ QueryMetadataInterface metadata,
+ CapabilitiesFinder capFinder,
+ AnalysisRecord analysisRecord, CommandContext context)
+ throws QueryPlannerException, QueryMetadataException, TeiidComponentException {
+ BatchedUpdateCommand batchedUpdateCommand = (BatchedUpdateCommand)command;
+ List childPlans = new ArrayList(batchedUpdateCommand.getUpdateCommands().size());
+ List updateCommands = batchedUpdateCommand.getUpdateCommands();
+ int numCommands = updateCommands.size();
+ List<VariableContext> allContexts = batchedUpdateCommand.getVariableContexts();
+ List<VariableContext> planContexts = null;
+ if (allContexts != null) {
+ planContexts = new ArrayList<VariableContext>(allContexts.size());
+ }
+ for (int commandIndex = 0; commandIndex < numCommands; commandIndex++) {
+ // Potentially the first command of a batch
+ Command updateCommand = (Command)updateCommands.get(commandIndex);
+ boolean commandWasBatched = false;
+ // If this command can be placed in a batch
+ if (isEligibleForBatching(updateCommand, metadata)) {
+ // Get the model ID. Subsequent and contiguous commands that update a group in this model are candidates for this batch
+ Object batchModelID = metadata.getModelID(getUpdatedGroup(updateCommand).getMetadataID());
+ String modelName = metadata.getFullName(batchModelID);
+ SourceCapabilities caps = capFinder.findCapabilities(modelName);
+ // Only attempt batching if the source supports batching
+ if (caps.supportsCapability(Capability.BATCHED_UPDATES)) {
+ // Start a new batch
+ List<Command> batch = new ArrayList<Command>();
+ List<VariableContext> contexts = new ArrayList<VariableContext>();
+ List<Boolean> shouldEvaluate = new ArrayList<Boolean>();
+ // This is the first command in a potential batch, so add it to the batch
+ batch.add(updateCommand);
+ if (allContexts != null) {
+ contexts.add(allContexts.get(commandIndex));
+ shouldEvaluate.add(Boolean.TRUE);
+ } else {
+ shouldEvaluate.add(EvaluatableVisitor.needsProcessingEvaluation(updateCommand));
+ }
+ // Find out if there are other commands called on the same physical model
+ // immediately and contiguously after this one
+ batchLoop: for (int batchIndex = commandIndex+1; batchIndex < numCommands; batchIndex++) {
+ Command batchingCandidate = (Command)updateCommands.get(batchIndex);
+ // If this command updates the same model, and is eligible for batching, add it to the batch
+ if (canBeAddedToBatch(batchingCandidate, batchModelID, metadata, capFinder)) {
+ batch.add(batchingCandidate);
+ if (allContexts != null) {
+ contexts.add(allContexts.get(batchIndex));
+ shouldEvaluate.add(Boolean.TRUE);
+ } else {
+ shouldEvaluate.add(EvaluatableVisitor.needsProcessingEvaluation(batchingCandidate));
+ }
+ } else { // Otherwise, stop batching at this point. The next command may well be the start of a new batch
+ break batchLoop;
+ }
+ }
+ // If two or more contiguous commands made on the same model were found, then batch them
+ if (batch.size() > 1) {
+ ProjectNode projectNode = new ProjectNode(((IntegerID)idGenerator.create()).getValue());
+ // Create a BatchedUpdateNode that creates a batched request for the connector
+ BatchedUpdateNode batchNode = new BatchedUpdateNode(((IntegerID)idGenerator.create()).getValue(),
+ batch, contexts, shouldEvaluate,
+ modelName);
+ List symbols = batchedUpdateCommand.getProjectedSymbols();
+ projectNode.setSelectSymbols(symbols);
+ projectNode.setElements(symbols);
+
+ batchNode.setElements(symbols);
+
+ projectNode.addChild(batchNode);
+ // Add a new RelationalPlan that represents the plan for this batch.
+ childPlans.add(new RelationalPlan(projectNode));
+ if (planContexts != null) {
+ planContexts.add(new VariableContext());
+ }
+ // Skip those commands that were added to this batch
+ commandIndex += batch.size() - 1;
+ commandWasBatched = true;
+ }
+ }
+ }
+ if (!commandWasBatched) { // If the command wasn't batched, just add the plan for this command to the list of plans
+ Command cmd = (Command)batchedUpdateCommand.getUpdateCommands().get(commandIndex);
+ ProcessorPlan plan = cmd.getProcessorPlan();
+ if (plan == null) {
+ plan = QueryOptimizer.optimizePlan(cmd, metadata, idGenerator, capFinder, analysisRecord, context);
+ }
+ childPlans.add(plan);
+ if (allContexts != null) {
+ planContexts.add(allContexts.get(commandIndex));
+ }
+ }
+ }
+ return new BatchedUpdatePlan(childPlans, batchedUpdateCommand.getUpdateCommands().size(), planContexts);
+ }
+
+ /**
+ * Get the group being updated by the update command
+ * @param command an INSERT, UPDATE, DELETE or SELECT INTO command
+ * @return the group being updated
+ * @since 4.2
+ */
+ public static GroupSymbol getUpdatedGroup(Command command) {
+ int type = command.getType();
+ if (type == Command.TYPE_INSERT) {
+ return ((Insert)command).getGroup();
+ } else if (type == Command.TYPE_UPDATE) {
+ return ((Update)command).getGroup();
+ } else if (type == Command.TYPE_DELETE) {
+ return ((Delete)command).getGroup();
+ }
+ throw new TeiidRuntimeException(QueryExecPlugin.Util.getString("BatchedUpdatePlanner.unrecognized_command", command)); //$NON-NLS-1$
+ }
+
+ /**
+ * Returns whether a command can be placed in a connector batch
+ * @param command an update command
+ * @param metadata
+ * @return true if this command can be added to a batch; false otherwise
+ * @throws QueryMetadataException
+ * @throws TeiidComponentException
+ * @since 4.2
+ */
+ public static boolean isEligibleForBatching(Command command, QueryMetadataInterface metadata) throws QueryMetadataException, TeiidComponentException {
+ // If the command updates a physical group, it's eligible
+ return !metadata.isVirtualGroup(getUpdatedGroup(command).getMetadataID());
+ }
+
+ /**
+ * Returns whether a command can be placed in a given batch
+ * @param command an update command
+ * @param batchModelID the model ID for the batch concerned
+ * @param metadata
+ * @param capFinder
+ * @return true if this command can be place in a batch associated with the model ID; false otherwise
+ * @throws QueryMetadataException
+ * @throws TeiidComponentException
+ * @since 4.2
+ */
+ private static boolean canBeAddedToBatch(Command command, Object batchModelID, QueryMetadataInterface metadata, CapabilitiesFinder capFinder) throws QueryMetadataException, TeiidComponentException {
+ // If it's eligible ...
+ if (isEligibleForBatching(command, metadata)) {
+ Object modelID = metadata.getModelID(getUpdatedGroup(command).getMetadataID());
+
+ return CapabilitiesUtil.isSameConnector(modelID, batchModelID, metadata, capFinder);
+ }
+ return false;
+ }
+
+}
Modified: trunk/engine/src/main/java/org/teiid/query/optimizer/QueryOptimizer.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/query/optimizer/QueryOptimizer.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/query/optimizer/QueryOptimizer.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -36,7 +36,6 @@
import org.teiid.query.metadata.TempCapabilitiesFinder;
import org.teiid.query.metadata.TempMetadataAdapter;
import org.teiid.query.metadata.TempMetadataStore;
-import org.teiid.query.optimizer.batch.BatchedUpdatePlanner;
import org.teiid.query.optimizer.capabilities.CapabilitiesFinder;
import org.teiid.query.optimizer.proc.ProcedurePlanner;
import org.teiid.query.optimizer.relational.RelationalPlanner;
Modified: trunk/engine/src/main/java/org/teiid/query/optimizer/capabilities/SourceCapabilities.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/query/optimizer/capabilities/SourceCapabilities.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/query/optimizer/capabilities/SourceCapabilities.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -317,7 +317,8 @@
* @since 6.0.0 indicates support for where all
*/
REQUIRES_CRITERIA,
- INSERT_WITH_QUERYEXPRESSION
+ INSERT_WITH_QUERYEXPRESSION,
+ INSERT_WITH_ITERATOR
}
public enum Scope {
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 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/query/optimizer/relational/PlanToProcessConverter.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -192,8 +192,15 @@
pinode.setModelName(modelName);
processNode = pinode;
SourceCapabilities caps = capFinder.findCapabilities(modelName);
- pinode.setDoBatching(caps.supportsCapability(Capability.BATCHED_UPDATES));
- pinode.setDoBulkInsert(caps.supportsCapability(Capability.BULK_UPDATE));
+ if (caps.supportsCapability(Capability.INSERT_WITH_ITERATOR)) {
+ pinode.setMode(org.teiid.query.processor.relational.ProjectIntoNode.Mode.ITERATOR);
+ } else if (caps.supportsCapability(Capability.BULK_UPDATE)) {
+ pinode.setMode(org.teiid.query.processor.relational.ProjectIntoNode.Mode.BULK);
+ } else if (caps.supportsCapability(Capability.BATCHED_UPDATES)) {
+ pinode.setMode(org.teiid.query.processor.relational.ProjectIntoNode.Mode.BATCH);
+ } else {
+ pinode.setMode(org.teiid.query.processor.relational.ProjectIntoNode.Mode.SINGLE);
+ }
}
} catch(QueryMetadataException e) {
throw new TeiidComponentException(e);
Modified: trunk/engine/src/main/java/org/teiid/query/processor/BatchIterator.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/query/processor/BatchIterator.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/query/processor/BatchIterator.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -31,7 +31,6 @@
import org.teiid.core.TeiidComponentException;
import org.teiid.core.TeiidProcessingException;
import org.teiid.query.processor.BatchCollector.BatchProducer;
-import org.teiid.query.sql.symbol.Expression;
/**
@@ -87,7 +86,6 @@
return tuple;
}
- @Override
public int available() {
if (batch != null && batch.containsRow(getCurrentIndex())) {
return batch.getEndRow() - getCurrentIndex() + 1;
@@ -95,11 +93,6 @@
return 0;
}
- @Override
- public List<? extends Expression> getSchema() {
- return source.getOutputElements();
- }
-
public void setBuffer(TupleBuffer buffer, boolean saveOnMark) {
this.buffer = buffer;
this.saveOnMark = saveOnMark;
Modified: trunk/engine/src/main/java/org/teiid/query/processor/CollectionTupleSource.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/query/processor/CollectionTupleSource.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/query/processor/CollectionTupleSource.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -28,27 +28,22 @@
import java.util.List;
import org.teiid.common.buffer.TupleSource;
-import org.teiid.query.sql.lang.Command;
-import org.teiid.query.sql.symbol.SingleElementSymbol;
public class CollectionTupleSource implements TupleSource {
private Iterator<? extends List<?>> tuples;
- private List<? extends SingleElementSymbol> schema;
public static CollectionTupleSource createUpdateCountTupleSource(int count) {
- return new CollectionTupleSource(Arrays.asList(Arrays.asList(count)).iterator(), Command.getUpdateCommandSymbol());
+ return new CollectionTupleSource(Arrays.asList(Arrays.asList(count)).iterator());
}
- public static CollectionTupleSource createNullTupleSource(List<SingleElementSymbol> schema) {
- return new CollectionTupleSource(new ArrayList<List<Object>>(0).iterator(), schema);
+ public static CollectionTupleSource createNullTupleSource() {
+ return new CollectionTupleSource(new ArrayList<List<Object>>(0).iterator());
}
- public CollectionTupleSource(Iterator<? extends List<?>> tuples,
- List<? extends SingleElementSymbol> schema) {
+ public CollectionTupleSource(Iterator<? extends List<?>> tuples) {
this.tuples = tuples;
- this.schema = schema;
}
@Override
@@ -60,20 +55,8 @@
}
@Override
- public List<? extends SingleElementSymbol> getSchema() {
- return schema;
- }
-
- @Override
public void closeSource() {
}
- @Override
- public int available() {
- if (tuples.hasNext()) {
- return 1;
- }
- return 0;
- }
}
\ No newline at end of file
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 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/query/processor/proc/ProcedurePlan.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -65,7 +65,6 @@
import org.teiid.query.sql.symbol.Reference;
import org.teiid.query.sql.util.VariableContext;
import org.teiid.query.tempdata.TempTableStore;
-import org.teiid.query.tempdata.TempTableStore;
import org.teiid.query.util.CommandContext;
import org.teiid.query.util.ErrorMessageKeys;
@@ -300,7 +299,7 @@
}
if(lastTupleSource == null){
- return CollectionTupleSource.createNullTupleSource(null);
+ return CollectionTupleSource.createNullTupleSource();
}
return lastTupleSource;
}
@@ -558,7 +557,7 @@
CursorState cursorState = getCursorState(rsKey);
// get the schema from the tuple source
- List schema = cursorState.ts.getSchema();
+ List schema = cursorState.processor.getOutputElements();
return schema;
}
Modified: trunk/engine/src/main/java/org/teiid/query/processor/relational/DependentCriteriaProcessor.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/query/processor/relational/DependentCriteriaProcessor.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/query/processor/relational/DependentCriteriaProcessor.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -89,7 +89,7 @@
sortSymbols.add(dependentSetStates.get(i).valueExpression);
}
DependentValueSource originalVs = (DependentValueSource)dependentNode.getContext().getVariableContext().getGlobalValue(valueSource);
- this.sortUtility = new SortUtility(originalVs.getTupleBuffer().createIndexedTupleSource(), sortSymbols, sortDirection, Mode.DUP_REMOVE, dependentNode.getBufferManager(), dependentNode.getConnectionID());
+ this.sortUtility = new SortUtility(originalVs.getTupleBuffer().createIndexedTupleSource(), sortSymbols, sortDirection, Mode.DUP_REMOVE, dependentNode.getBufferManager(), dependentNode.getConnectionID(), originalVs.getTupleBuffer().getSchema());
}
dvs = new DependentValueSource(sortUtility.sort());
for (SetState setState : dependentSetStates) {
Modified: trunk/engine/src/main/java/org/teiid/query/processor/relational/DependentValueSource.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/query/processor/relational/DependentValueSource.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/query/processor/relational/DependentValueSource.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -65,7 +65,7 @@
IndexedTupleSource its = buffer.createIndexedTupleSource();
int index = 0;
if (valueExpression != null) {
- index = its.getSchema().indexOf(valueExpression);
+ index = buffer.getSchema().indexOf(valueExpression);
Assertion.assertTrue(index != -1);
}
return new TupleSourceValueIterator(its, index);
@@ -83,10 +83,10 @@
IndexedTupleSource its = buffer.createIndexedTupleSource();
int index = 0;
if (valueExpression != null) {
- index = its.getSchema().indexOf(valueExpression);
+ index = buffer.getSchema().indexOf(valueExpression);
}
Assertion.assertTrue(index != -1);
- if (((SingleElementSymbol)its.getSchema().get(index)).getType() == DataTypeManager.DefaultDataClasses.BIG_DECIMAL) {
+ if (((SingleElementSymbol)buffer.getSchema().get(index)).getType() == DataTypeManager.DefaultDataClasses.BIG_DECIMAL) {
result = new TreeSet<Object>();
} else {
result = new HashSet<Object>();
Modified: trunk/engine/src/main/java/org/teiid/query/processor/relational/GroupingNode.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/query/processor/relational/GroupingNode.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/query/processor/relational/GroupingNode.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -306,22 +306,10 @@
}
@Override
- public List<Expression> getSchema() {
- return collectedExpressions;
- }
-
- @Override
public void closeSource() {
}
- @Override
- public int available() {
- if (sourceBatch != null) {
- return sourceBatch.getEndRow() - sourceRow + 1;
- }
- return 0;
- }
};
}
@@ -334,7 +322,7 @@
} else {
this.sortUtility = new SortUtility(getCollectionTupleSource(), sortElements,
sortTypes, removeDuplicates?Mode.DUP_REMOVE_SORT:Mode.SORT, getBufferManager(),
- getConnectionID());
+ getConnectionID(), collectedExpressions);
this.phase = SORT;
}
}
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 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/query/processor/relational/ProjectIntoNode.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -32,7 +32,9 @@
import org.teiid.client.plan.PlanNode;
import org.teiid.common.buffer.BlockedException;
import org.teiid.common.buffer.TupleBatch;
+import org.teiid.common.buffer.TupleBuffer;
import org.teiid.common.buffer.TupleSource;
+import org.teiid.common.buffer.BufferManager.TupleSourceType;
import org.teiid.core.TeiidComponentException;
import org.teiid.core.TeiidProcessingException;
import org.teiid.query.sql.lang.BatchedUpdateCommand;
@@ -45,6 +47,10 @@
public class ProjectIntoNode extends RelationalNode {
+ public enum Mode {
+ BATCH, BULK, ITERATOR, SINGLE
+ }
+
private static int REQUEST_CREATION = 1;
private static int RESPONSE_PROCESSING = 2;
@@ -52,8 +58,7 @@
private GroupSymbol intoGroup;
private List intoElements;
private String modelName;
- private boolean doBatching = false;
- private boolean doBulkInsert = false;
+ private Mode mode;
// Processing state
private int batchRow = 1;
@@ -61,7 +66,9 @@
private int phase = REQUEST_CREATION;
private int requestsRegistered = 0;
private int tupleSourcesProcessed = 0;
+ private boolean sourceDone;
+ private TupleBuffer buffer;
private TupleBatch currentBatch;
private TupleSource tupleSource;
@@ -78,6 +85,7 @@
this.tupleSourcesProcessed = 0;
this.requestsRegistered = 0;
this.currentBatch=null;
+ this.sourceDone=false;
}
public void setIntoGroup(GroupSymbol group) {
@@ -107,28 +115,45 @@
checkExitConditions();
- /* If we don't have a batch to work with yet, or
- * if this is not the last batch and we've reached the end of this one,
- * then get the next batch.
+ /* If we don't have a batch to work, get the next
*/
-
- if (currentBatch == null || (!currentBatch.getTerminationFlag() && this.batchRow > currentBatch.getEndRow())) {
+ if (currentBatch == null) {
+ if (sourceDone) {
+ phase = RESPONSE_PROCESSING;
+ break;
+ }
currentBatch = getChildren()[0].nextBatch(); // can throw BlockedException
+ sourceDone = currentBatch.getTerminationFlag();
this.batchRow = currentBatch.getBeginRow();
//normally we would want to skip a 0 sized batch, but it typically represents the terminal batch
//and for implicit temp tables we need to issue an empty insert
- if(currentBatch.getRowCount() == 0 && (!this.intoGroup.isImplicitTempGroupSymbol() || !currentBatch.getTerminationFlag())) {
- continue;
+ if(currentBatch.getRowCount() == 0
+ && (!currentBatch.getTerminationFlag() || mode != Mode.ITERATOR)) {
+ currentBatch = null;
+ continue;
}
- } else if (currentBatch.getTerminationFlag() && this.batchRow > currentBatch.getEndRow()) {
- phase = RESPONSE_PROCESSING;
- break;
- }
+ }
int batchSize = currentBatch.getRowCount();
-
- if (doBulkInsert) {
+ int requests = 1;
+ switch (mode) {
+ case ITERATOR:
+ if (buffer == null) {
+ buffer = getBufferManager().createTupleBuffer(intoElements, getConnectionID(), TupleSourceType.PROCESSOR);
+ }
+ buffer.addTupleBatch(currentBatch, true);
+ if (currentBatch.getTerminationFlag() && (buffer.getRowCount() != 0 || intoGroup.isImplicitTempGroupSymbol())) {
+ Insert insert = new Insert(intoGroup, intoElements, null);
+ buffer.close();
+ insert.setTupleSource(buffer.createIndexedTupleSource(true));
+ // Register insert command against source
+ registerRequest(insert);
+ } else {
+ requests = 0;
+ }
+ break;
+ case BULK:
//convert to multivalued parameter
List<Constant> parameters = new ArrayList<Constant>(intoElements.size());
for (int i = 0; i < intoElements.size(); i++) {
@@ -145,26 +170,32 @@
Insert insert = new Insert(intoGroup, intoElements, parameters);
// Register insert command against source
registerRequest(insert);
- } else if (doBatching) {
+ break;
+ case BATCH:
// Register batched update command against source
int endRow = currentBatch.getEndRow();
List rows = new ArrayList(endRow-batchRow);
for(int rowNum = batchRow; rowNum <= endRow; rowNum++) {
- Insert insert = new Insert( intoGroup,
+ insert = new Insert( intoGroup,
intoElements,
convertValuesToConstants(currentBatch.getTuple(rowNum), intoElements));
rows.add( insert );
}
registerRequest(new BatchedUpdateCommand( rows ));
- } else {
+ break;
+ case SINGLE:
batchSize = 1;
// Register insert command against source
// Defect 16036 - submit a new INSERT command to the DataManager.
registerRequest(new Insert(intoGroup, intoElements, convertValuesToConstants(currentBatch.getTuple(batchRow), intoElements)));
}
+
this.batchRow += batchSize;
- this.requestsRegistered++;
+ if (batchRow > currentBatch.getEndRow()) {
+ currentBatch = null;
+ }
+ this.requestsRegistered+=requests;
}
checkExitConditions();
@@ -197,7 +228,10 @@
}
private void closeRequest() {
-
+ if (this.buffer != null) {
+ this.buffer.remove();
+ this.buffer = null;
+ }
if (this.tupleSource != null) {
tupleSource.closeSource();
this.tupleSource = null;
@@ -216,8 +250,7 @@
clonedNode.intoGroup = intoGroup;
clonedNode.intoElements = intoElements;
clonedNode.modelName = this.modelName;
- clonedNode.doBatching = this.doBatching;
- clonedNode.doBulkInsert = this.doBulkInsert;
+ clonedNode.mode = this.mode;
return clonedNode;
}
@@ -242,15 +275,15 @@
constants.add(new Constant(values.get(i),type));
}
return constants;
- }
+ }
+
+ public Mode getMode() {
+ return mode;
+ }
- public void setDoBatching(boolean doBatching) {
- this.doBatching = doBatching;
- }
-
- public void setDoBulkInsert(boolean doBulkInsert) {
- this.doBulkInsert = doBulkInsert;
- }
+ public void setMode(Mode mode) {
+ this.mode = mode;
+ }
public boolean isTempGroupInsert() {
return intoGroup.isTempGroupSymbol();
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 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/query/processor/relational/RelationalPlan.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -34,6 +34,7 @@
import org.teiid.core.TeiidProcessingException;
import org.teiid.query.processor.ProcessorDataManager;
import org.teiid.query.processor.ProcessorPlan;
+import org.teiid.query.processor.relational.ProjectIntoNode.Mode;
import org.teiid.query.sql.lang.QueryCommand;
import org.teiid.query.util.CommandContext;
@@ -157,7 +158,12 @@
}
return false;
}
- if (node instanceof AccessNode) {
+ if (node instanceof ProjectIntoNode) {
+ if (((ProjectIntoNode)node).getMode() == Mode.ITERATOR) {
+ return false;
+ }
+ return true;
+ } else if (node instanceof AccessNode) {
return false;
}
if (transactionalReads) {
Modified: trunk/engine/src/main/java/org/teiid/query/processor/relational/SortNode.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/query/processor/relational/SortNode.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/query/processor/relational/SortNode.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -91,7 +91,7 @@
private void sortPhase() throws BlockedException, TeiidComponentException, TeiidProcessingException {
if (this.sortUtility == null) {
this.sortUtility = new SortUtility(new BatchIterator(getChildren()[0]), items, this.mode, getBufferManager(),
- getConnectionID());
+ getConnectionID(), getChildren()[0].getElements());
}
this.output = this.sortUtility.sort();
if (this.outputTs == null) {
Modified: trunk/engine/src/main/java/org/teiid/query/processor/relational/SortUtility.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/query/processor/relational/SortUtility.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/query/processor/relational/SortUtility.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -108,12 +108,12 @@
private Collection<List<?>> workingTuples;
public SortUtility(TupleSource sourceID, List<OrderByItem> items, Mode mode, BufferManager bufferMgr,
- String groupName) {
+ String groupName, List schema) {
this.source = sourceID;
this.mode = mode;
this.bufferManager = bufferMgr;
this.groupName = groupName;
- this.schema = this.source.getSchema();
+ this.schema = schema;
this.schemaSize = bufferManager.getSchemaSize(this.schema);
int distinctIndex = items != null? items.size() - 1:0;
List<Expression> sortElements = null;
@@ -153,8 +153,8 @@
}
public SortUtility(TupleSource ts, List expressions, List<Boolean> types,
- Mode mode, BufferManager bufferManager, String connectionID) {
- this(ts, new OrderBy(expressions, types).getOrderByItems(), mode, bufferManager, connectionID);
+ Mode mode, BufferManager bufferManager, String connectionID, List schema) {
+ this(ts, new OrderBy(expressions, types).getOrderByItems(), mode, bufferManager, connectionID, schema);
}
public boolean isDone() {
Modified: trunk/engine/src/main/java/org/teiid/query/processor/relational/SortingFilter.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/query/processor/relational/SortingFilter.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/query/processor/relational/SortingFilter.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -134,7 +134,7 @@
// Sort
if (sortUtility == null) {
- sortUtility = new SortUtility(collectionBuffer.createIndexedTupleSource(), sortItems, removeDuplicates?Mode.DUP_REMOVE_SORT:Mode.SORT, mgr, groupName);
+ sortUtility = new SortUtility(collectionBuffer.createIndexedTupleSource(), sortItems, removeDuplicates?Mode.DUP_REMOVE_SORT:Mode.SORT, mgr, groupName, collectionBuffer.getSchema());
}
TupleBuffer sorted = sortUtility.sort();
sorted.setForwardOnly(true);
Modified: trunk/engine/src/main/java/org/teiid/query/processor/relational/SourceState.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/query/processor/relational/SourceState.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/query/processor/relational/SourceState.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -187,7 +187,7 @@
ts = new BatchIterator(this.source);
}
this.sortUtility = new SortUtility(ts, expressions, Collections.nCopies(expressions.size(), OrderBy.ASC),
- sortOption == SortOption.SORT_DISTINCT?Mode.DUP_REMOVE_SORT:Mode.SORT, this.source.getBufferManager(), this.source.getConnectionID());
+ sortOption == SortOption.SORT_DISTINCT?Mode.DUP_REMOVE_SORT:Mode.SORT, this.source.getBufferManager(), this.source.getConnectionID(), source.getElements());
this.markDistinct(sortOption == SortOption.SORT_DISTINCT && expressions.size() == this.getOuterVals().size());
}
this.buffer = sortUtility.sort();
Modified: trunk/engine/src/main/java/org/teiid/query/sql/lang/Insert.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/query/sql/lang/Insert.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/query/sql/lang/Insert.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -29,6 +29,7 @@
import java.util.List;
import java.util.Map;
+import org.teiid.common.buffer.TupleSource;
import org.teiid.core.util.EquivalenceUtil;
import org.teiid.core.util.HashCodeUtil;
import org.teiid.query.sql.LanguageVisitor;
@@ -38,6 +39,7 @@
import org.teiid.query.sql.symbol.Expression;
import org.teiid.query.sql.symbol.GroupSymbol;
import org.teiid.query.sql.symbol.SingleElementSymbol;
+import org.teiid.query.sql.util.ValueIterator;
/**
@@ -52,10 +54,12 @@
/** list of column variables, null = all columns */
private List variables = new LinkedList();
- /** List of Expressions, required */
+ /** List of Expressions */
private List values = new LinkedList();
private QueryCommand queryExpression;
+
+ private TupleSource tupleSource;
// =========================================================================
// C O N S T R U C T O R S
@@ -298,5 +302,13 @@
public boolean areResultsCachable() {
return false;
}
+
+ public void setTupleSource(TupleSource tupleSource) {
+ this.tupleSource = tupleSource;
+ }
+
+ public TupleSource getTupleSource() {
+ return tupleSource;
+ }
}
Modified: trunk/engine/src/main/java/org/teiid/query/sql/lang/Query.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/query/sql/lang/Query.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/query/sql/lang/Query.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -434,7 +434,7 @@
static boolean areResultsCachable(List projectedSymbols) {
for(int i=0; i<projectedSymbols.size(); i++){
SingleElementSymbol projectedSymbol = (SingleElementSymbol)projectedSymbols.get(i);
- if(DataTypeManager.isLOB(projectedSymbol.getType())) {
+ if(DataTypeManager.isLOB(projectedSymbol.getType()) || projectedSymbol.getType() == DataTypeManager.DefaultDataClasses.OBJECT) {
return false;
}
}
Modified: trunk/engine/src/main/java/org/teiid/query/sql/visitor/SQLStringVisitor.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/query/sql/visitor/SQLStringVisitor.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/query/sql/visitor/SQLStringVisitor.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -403,6 +403,9 @@
if ( obj.getQueryExpression() != null ) {
parts.add(registerNode(obj.getQueryExpression()));
+ } else if (obj.getTupleSource() != null) {
+ parts.add(VALUES);
+ parts.add(" (...)"); //$NON-NLS-1$
} else {
parts.add(VALUES);
parts.add(" ("); //$NON-NLS-1$
Modified: trunk/engine/src/main/java/org/teiid/query/tempdata/TempTable.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/query/tempdata/TempTable.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/query/tempdata/TempTable.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -37,6 +37,7 @@
import org.teiid.common.buffer.TupleBrowser;
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;
@@ -65,29 +66,75 @@
*/
class TempTable {
- private final class QueryTupleSource extends TupleBrowserTupleSource {
+ private final class InsertUpdateProcessor extends UpdateProcessor {
+
+ private boolean addRowId;
+ private int[] indexes;
+
+ private InsertUpdateProcessor(TupleSource ts, boolean addRowId, int[] indexes)
+ throws TeiidComponentException {
+ super(null, ts);
+ this.addRowId = addRowId;
+ this.indexes = indexes;
+ }
+
+ @Override
+ protected void tuplePassed(List tuple) throws BlockedException,
+ TeiidComponentException, TeiidProcessingException {
+ if (indexes != null) {
+ List<Object> newTuple = new ArrayList<Object>(columns.size());
+ if (addRowId) {
+ newTuple.add(rowId.getAndIncrement());
+ }
+ for (int i = 0; i < indexes.length; i++) {
+ if (indexes[i] == -1) {
+ newTuple.add(null);
+ } else {
+ newTuple.add(tuple.get(indexes[i]));
+ }
+ }
+ tuple = newTuple;
+ } else if (addRowId) {
+ tuple = new ArrayList<Object>(tuple);
+ tuple.add(0, rowId.getAndIncrement());
+ }
+ currentTuple = tuple;
+ insertTuple(tuple);
+ }
+
+ @Override
+ protected void undo(List tuple) throws TeiidComponentException,
+ TeiidProcessingException {
+ deleteTuple(tuple);
+ }
+ }
+
+ private final class QueryTupleSource implements TupleSource {
private final Evaluator eval;
- private final List<SingleElementSymbol> projectedCols;
private final Criteria condition;
private final boolean project;
private final int[] indexes;
+ private int reserved;
+ private TupleBrowser browser;
private QueryTupleSource(TupleBrowser browser, Map map,
List<SingleElementSymbol> projectedCols, Criteria condition) {
- super(browser);
+ this.browser = browser;
this.indexes = RelationalNode.getProjectionIndexes(map, projectedCols);
this.eval = new Evaluator(map, null, null);
- this.projectedCols = projectedCols;
this.condition = condition;
this.project = shouldProject();
+ reserved = reserveBuffers();
}
@Override
public List<?> nextTuple() throws TeiidComponentException,
TeiidProcessingException {
for (;;) {
- List<?> next = super.nextTuple();
+ List<?> next = browser.nextTuple();
if (next == null) {
+ bm.releaseBuffers(reserved);
+ reserved = 0;
return null;
}
if (rowId != null) {
@@ -103,6 +150,13 @@
}
}
+ @Override
+ public void closeSource() {
+ bm.releaseBuffers(reserved);
+ reserved = 0;
+ browser.closeSource();
+ }
+
private boolean shouldProject() {
if (indexes.length == getColumns().size()) {
for (int i = 0; i < indexes.length; i++) {
@@ -115,48 +169,15 @@
return true;
}
- @Override
- public List<? extends Expression> getSchema() {
- return projectedCols;
- }
}
- private class TupleBrowserTupleSource implements TupleSource {
- private final TupleBrowser browser;
-
- private TupleBrowserTupleSource(TupleBrowser browser) {
- this.browser = browser;
- }
-
- @Override
- public List<?> nextTuple() throws TeiidComponentException,
- TeiidProcessingException {
- return browser.next();
- }
-
- @Override
- public List<? extends Expression> getSchema() {
- return columns;
- }
-
- @Override
- public void closeSource() {
-
- }
-
- @Override
- public int available() {
- return 0;
- }
- }
-
private abstract class UpdateProcessor {
private TupleSource ts;
protected final Map lookup;
protected final Evaluator eval;
private final Criteria crit;
protected int updateCount = 0;
- private List currentTuple;
+ protected List currentTuple;
protected TupleBuffer undoLog;
@@ -169,6 +190,7 @@
}
int process() throws ExpressionEvaluationException, TeiidComponentException, TeiidProcessingException {
+ int reserved = reserveBuffers();
boolean success = false;
try {
while (currentTuple != null || (currentTuple = ts.nextTuple()) != null) {
@@ -179,9 +201,12 @@
}
currentTuple = null;
}
+ bm.releaseBuffers(reserved);
+ reserved = 0;
success();
success = true;
} finally {
+ bm.releaseBuffers(reserved);
if (!success) {
TupleSource undoTs = undoLog.createIndexedTupleSource();
List<?> tuple = null;
@@ -197,7 +222,7 @@
}
return updateCount;
}
-
+
@SuppressWarnings("unused")
void success() throws TeiidComponentException, ExpressionEvaluationException, TeiidProcessingException {}
@@ -218,6 +243,9 @@
private BufferManager bm;
private String sessionID;
private TempMetadataID tid;
+
+ private int keyBatchSize;
+ private int leafBatchSize;
TempTable(TempMetadataID tid, BufferManager bm, List<ElementSymbol> columns, int primaryKeyLength, String sessionID) {
this.tid = tid;
@@ -233,7 +261,13 @@
}
this.columns = columns;
this.sessionID = sessionID;
+ this.keyBatchSize = bm.getSchemaSize(columns);
+ this.leafBatchSize = bm.getSchemaSize(columns.subList(0, primaryKeyLength));
}
+
+ private int reserveBuffers() {
+ return bm.reserveBuffers(leafBatchSize + (tree.getHeight() - 1)*keyBatchSize, BufferReserveMode.WAIT);
+ }
public TupleSource createTupleSource(final List<SingleElementSymbol> projectedCols, final Criteria condition, OrderBy orderBy) throws TeiidComponentException, TeiidProcessingException {
Map map = RelationalNode.createLookupMap(getColumns());
@@ -282,21 +316,24 @@
TupleBrowser browser = createTupleBrower(condition, direction);
TupleSource ts = new QueryTupleSource(browser, map, projectedCols, condition);
-
- TupleBuffer tb = null;
- if (!orderByUsingIndex && orderBy != null) {
- SortUtility sort = new SortUtility(ts, orderBy.getOrderByItems(), Mode.SORT, bm, sessionID);
- tb = sort.sort();
- } else {
- tb = bm.createTupleBuffer(projectedCols, sessionID, TupleSourceType.PROCESSOR);
- List next = null;
- while ((next = ts.nextTuple()) != null) {
- tb.addTuple(next);
+ try {
+ TupleBuffer tb = null;
+ if (!orderByUsingIndex && orderBy != null) {
+ SortUtility sort = new SortUtility(ts, orderBy.getOrderByItems(), Mode.SORT, bm, sessionID, projectedCols);
+ tb = sort.sort();
+ } else {
+ tb = bm.createTupleBuffer(projectedCols, sessionID, TupleSourceType.PROCESSOR);
+ List next = null;
+ while ((next = ts.nextTuple()) != null) {
+ tb.addTuple(next);
+ }
}
+ tb.close();
+ return tb.createIndexedTupleSource(true);
+ } finally {
+ //ensure the buffers get released
+ ts.closeSource();
}
- tb.close();
- tb.setForwardOnly(true);
- return tb.createIndexedTupleSource(true);
}
private TupleBrowser createTupleBrower(Criteria condition, boolean direction) throws TeiidComponentException {
@@ -353,23 +390,15 @@
return columns;
}
- public TupleSource insert(List<List<Object>> tuples) throws TeiidComponentException, ExpressionEvaluationException, TeiidProcessingException {
- UpdateProcessor up = new UpdateProcessor(null, new CollectionTupleSource(tuples.iterator(), getColumns())) {
-
- protected void tuplePassed(List tuple)
- throws BlockedException, TeiidComponentException, TeiidProcessingException {
- if (rowId != null) {
- tuple.add(0, rowId.getAndIncrement());
- }
- insertTuple(tuple);
- }
-
- @Override
- protected void undo(List tuple) throws TeiidComponentException {
- deleteTuple(tuple);
- }
-
- };
+ public TupleSource insert(TupleSource tuples, final List<ElementSymbol> variables) throws TeiidComponentException, ExpressionEvaluationException, TeiidProcessingException {
+ List<ElementSymbol> cols = getColumns();
+ final int[] indexes = new int[cols.size()];
+ boolean shouldProject = false;
+ for (int i = 0; i < cols.size(); i++) {
+ indexes[i] = variables.indexOf(cols.get(i));
+ shouldProject |= (indexes[i] != i);
+ }
+ UpdateProcessor up = new InsertUpdateProcessor(tuples, rowId != null, shouldProject?indexes:null);
int updateCount = up.process();
tid.setCardinality(tree.getRowCount());
return CollectionTupleSource.createUpdateCountTupleSource(updateCount);
@@ -378,7 +407,7 @@
public TupleSource update(Criteria crit, final SetClauseList update) throws TeiidComponentException, ExpressionEvaluationException, TeiidProcessingException {
final boolean primaryKeyChangePossible = canChangePrimaryKey(update);
final TupleBrowser browser = createTupleBrower(crit, OrderBy.ASC);
- UpdateProcessor up = new UpdateProcessor(crit, new TupleBrowserTupleSource(browser)) {
+ UpdateProcessor up = new UpdateProcessor(crit, browser) {
protected TupleBuffer changeSet;
protected UpdateProcessor changeSetProcessor;
@@ -418,20 +447,7 @@
//changeSet contains possible updates
if (primaryKeyChangePossible) {
if (changeSetProcessor == null) {
- changeSetProcessor = new UpdateProcessor(null, changeSet.createIndexedTupleSource(true)) {
- @Override
- protected void tuplePassed(List tuple) throws BlockedException,
- TeiidComponentException, TeiidProcessingException {
- insertTuple(tuple);
- }
-
- @Override
- protected void undo(List tuple) throws TeiidComponentException,
- TeiidProcessingException {
- deleteTuple(tuple);
- }
-
- };
+ changeSetProcessor = new InsertUpdateProcessor(changeSet.createIndexedTupleSource(true), false, null);
}
changeSetProcessor.process(); //when this returns, we're up to date
}
@@ -466,7 +482,7 @@
public TupleSource delete(Criteria crit) throws TeiidComponentException, ExpressionEvaluationException, TeiidProcessingException {
final TupleBrowser browser = createTupleBrower(crit, OrderBy.ASC);
- UpdateProcessor up = new UpdateProcessor(crit, new TupleBrowserTupleSource(browser)) {
+ UpdateProcessor up = new UpdateProcessor(crit, browser) {
@Override
protected void tuplePassed(List tuple)
throws ExpressionEvaluationException,
@@ -502,5 +518,5 @@
throw new AssertionError("Update failed"); //$NON-NLS-1$
}
}
-
+
}
\ No newline at end of file
Modified: trunk/engine/src/main/java/org/teiid/query/tempdata/TempTableStore.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/query/tempdata/TempTableStore.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/main/java/org/teiid/query/tempdata/TempTableStore.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -23,15 +23,14 @@
package org.teiid.query.tempdata;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import org.teiid.api.exception.query.ExpressionEvaluationException;
import org.teiid.api.exception.query.QueryProcessingException;
-import org.teiid.common.buffer.BlockedException;
import org.teiid.common.buffer.BufferManager;
import org.teiid.common.buffer.TupleSource;
import org.teiid.core.TeiidComponentException;
@@ -52,7 +51,6 @@
import org.teiid.query.sql.lang.Query;
import org.teiid.query.sql.lang.Update;
import org.teiid.query.sql.navigator.PostOrderNavigator;
-import org.teiid.query.sql.symbol.Constant;
import org.teiid.query.sql.symbol.ElementSymbol;
import org.teiid.query.sql.symbol.Expression;
import org.teiid.query.sql.symbol.GroupSymbol;
@@ -142,11 +140,16 @@
final String groupKey = group.getNonCorrelationName().toUpperCase();
final TempTable table = getTempTable(groupKey, command);
if (command instanceof Insert) {
- List<List<Object>> tuples = getBulkRows((Insert)command, table.getColumns());
- if (tuples.isEmpty()) {
- return CollectionTupleSource.createUpdateCountTupleSource(0);
+ Insert insert = (Insert)command;
+ TupleSource ts = insert.getTupleSource();
+ if (ts == null) {
+ List<Object> values = new ArrayList<Object>(insert.getValues().size());
+ for (Expression expr : (List<Expression>)insert.getValues()) {
+ values.add(Evaluator.evaluate(expr));
+ }
+ ts = new CollectionTupleSource(Arrays.asList(values).iterator());
}
- return table.insert(tuples);
+ return table.insert(ts, insert.getVariables());
}
if (command instanceof Update) {
final Update update = (Update)command;
@@ -213,36 +216,6 @@
return groupToTupleSourceID.get(tempTableID);
}
- public static List<List<Object>> getBulkRows(Insert insert, List<ElementSymbol> elements) throws ExpressionEvaluationException, BlockedException, TeiidComponentException {
- int bulkRowCount = 1;
- if (insert.isBulk()) {
- Constant c = (Constant)insert.getValues().get(0);
- bulkRowCount = ((List<?>)c.getValue()).size();
- }
-
- List<List<Object>> tuples = new ArrayList<List<Object>>(bulkRowCount);
-
- for (int row = 0; row < bulkRowCount; row++) {
- List<Object> currentRow = new ArrayList<Object>(insert.getValues().size());
- for (ElementSymbol symbol : elements) {
- int index = insert.getVariables().indexOf(symbol);
- Object value = null;
- if (index != -1) {
- if (insert.isBulk()) {
- Constant multiValue = (Constant)insert.getValues().get(index);
- value = ((List<?>)multiValue.getValue()).get(row);
- } else {
- Expression expr = (Expression)insert.getValues().get(index);
- value = Evaluator.evaluate(expr);
- }
- }
- currentRow.add(value);
- }
- tuples.add(currentRow);
- }
- return tuples;
- }
-
public boolean hasTempTable(Command command){
switch (command.getType()) {
case Command.TYPE_INSERT:
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 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/test/java/org/teiid/dqp/internal/process/TestDQPCore.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -36,7 +36,6 @@
import org.teiid.api.exception.query.QueryResolverException;
import org.teiid.client.RequestMessage;
import org.teiid.client.ResultsMessage;
-import org.teiid.common.queue.FakeWorkManager;
import org.teiid.dqp.internal.datamgr.ConnectorManagerRepository;
import org.teiid.dqp.internal.datamgr.FakeTransactionService;
import org.teiid.dqp.service.AutoGenDataService;
Modified: trunk/engine/src/test/java/org/teiid/query/optimizer/TestOptimizer.java
===================================================================
--- trunk/engine/src/test/java/org/teiid/query/optimizer/TestOptimizer.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/test/java/org/teiid/query/optimizer/TestOptimizer.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -6413,12 +6413,26 @@
FakeMetadataFacade metadata = FakeMetadataFactory.example1Cached();
ProcessorPlan plan = helpPlan(sql, metadata, new String[] {"SELECT DISTINCT g_0.e1 FROM pm1.g1 AS g_0"}, ComparisonMode.EXACT_COMMAND_STRING); //$NON-NLS-1$
+ //no txn required, since an interated insert is used
+ assertFalse(plan.requiresTransaction(false));
checkNodeTypes(plan, FULL_PUSHDOWN);
checkNodeTypes(plan, new int[] {1}, new Class[] {ProjectIntoNode.class});
}
+ @Test public void testInsertQueryExpression() throws Exception {
+ String sql = "insert into pm1.g1 (e1) select e1 from pm1.g2"; //$NON-NLS-1$
+
+ FakeMetadataFacade metadata = FakeMetadataFactory.example1Cached();
+
+ ProcessorPlan plan = helpPlan(sql, metadata, new String[] {"SELECT g_0.e1 FROM pm1.g2 AS g_0"}, ComparisonMode.EXACT_COMMAND_STRING); //$NON-NLS-1$
+ //requires a txn, since an non pushdown/iterated insert is used
+ assertTrue(plan.requiresTransaction(false));
+
+ checkNodeTypes(plan, new int[] {1}, new Class[] {ProjectIntoNode.class});
+ }
+
/**
* previously the subqueries were being pushed too far and then not having the appropriate correlated references
*/
Modified: trunk/engine/src/test/java/org/teiid/query/processor/FakeTupleSource.java
===================================================================
--- trunk/engine/src/test/java/org/teiid/query/processor/FakeTupleSource.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/test/java/org/teiid/query/processor/FakeTupleSource.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -45,8 +45,6 @@
//the first time nextTuple is called, it will throws BlockedExceptiom
private boolean blockOnce;
- private boolean exceptionOnClose;
-
public FakeTupleSource(List elements, List[] tuples) {
this.elements = elements;
this.tuples = tuples;
@@ -113,13 +111,4 @@
this.blockOnce = true;
}
- public void setExceptionOnClose(boolean exceptionOnClose) {
- this.exceptionOnClose = exceptionOnClose;
- }
-
- @Override
- public int available() {
- return 0;
- }
-
}
Modified: trunk/engine/src/test/java/org/teiid/query/processor/TestCollectionTupleSource.java
===================================================================
--- trunk/engine/src/test/java/org/teiid/query/processor/TestCollectionTupleSource.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/test/java/org/teiid/query/processor/TestCollectionTupleSource.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -29,8 +29,6 @@
import java.util.List;
import org.junit.Test;
-import org.teiid.query.processor.CollectionTupleSource;
-import org.teiid.query.sql.lang.Command;
import org.teiid.query.sql.symbol.ElementSymbol;
import org.teiid.query.sql.symbol.SingleElementSymbol;
@@ -42,11 +40,8 @@
List<SingleElementSymbol> elements = new ArrayList<SingleElementSymbol>();
elements.add(new ElementSymbol("x")); //$NON-NLS-1$
elements.add(new ElementSymbol("y")); //$NON-NLS-1$
- CollectionTupleSource nts = CollectionTupleSource.createNullTupleSource(elements);
+ CollectionTupleSource nts = CollectionTupleSource.createNullTupleSource();
- // Check schema
- assertEquals("Didn't get expected schema", elements, nts.getSchema()); //$NON-NLS-1$
-
// Walk it and get no data
List tuple = nts.nextTuple();
nts.closeSource();
@@ -57,9 +52,6 @@
@Test public void testUpdateCountSource() {
CollectionTupleSource nts = CollectionTupleSource.createUpdateCountTupleSource(5);
- // Check schema
- assertEquals("Didn't get expected schema", Command.getUpdateCommandSymbol(), nts.getSchema()); //$NON-NLS-1$
-
// Walk it and get no data
List tuple = nts.nextTuple();
nts.closeSource();
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 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/test/java/org/teiid/query/processor/relational/TestBatchedUpdateNode.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -39,7 +39,6 @@
import org.teiid.query.metadata.QueryMetadataInterface;
import org.teiid.query.optimizer.batch.TestBatchedUpdatePlanner;
import org.teiid.query.processor.ProcessorDataManager;
-import org.teiid.query.processor.relational.BatchedUpdateNode;
import org.teiid.query.sql.lang.BatchedUpdateCommand;
import org.teiid.query.sql.lang.Command;
import org.teiid.query.sql.visitor.EvaluatableVisitor;
@@ -233,7 +232,6 @@
this.numCommands = numCommands;
}
public void closeSource() {}
- public List getSchema() {return null;}
public List nextTuple() throws TeiidComponentException {
if (first) {
first = false;
@@ -245,10 +243,6 @@
return null;
}
- @Override
- public int available() {
- return 0;
- }
}
}
Modified: trunk/engine/src/test/java/org/teiid/query/processor/relational/TestGroupingNode.java
===================================================================
--- trunk/engine/src/test/java/org/teiid/query/processor/relational/TestGroupingNode.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/test/java/org/teiid/query/processor/relational/TestGroupingNode.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -57,7 +57,7 @@
public class TestGroupingNode {
- public static TupleSource createTupleSource1() {
+ public static FakeTupleSource createTupleSource1() {
List<ElementSymbol> symbols = new ArrayList<ElementSymbol>();
symbols.add(new ElementSymbol("col1")); //$NON-NLS-1$
symbols.get(0).setType(DataTypeManager.DefaultDataClasses.INTEGER);
@@ -90,7 +90,7 @@
List[] expected, ProcessorDataManager dataMgr) throws TeiidComponentException,
BlockedException,
TeiidProcessingException {
- TupleSource dataSource = createTupleSource1();
+ FakeTupleSource dataSource = createTupleSource1();
helpProcess(mgr, node, context, expected, dataSource, dataMgr);
}
@@ -98,7 +98,7 @@
GroupingNode node,
CommandContext context,
List[] expected,
- TupleSource dataSource, ProcessorDataManager dataMgr) throws TeiidComponentException,
+ FakeTupleSource dataSource, ProcessorDataManager dataMgr) throws TeiidComponentException,
BlockedException,
TeiidProcessingException {
RelationalNode dataNode = new FakeRelationalNode(0, dataSource, mgr.getProcessorBatchSize());
@@ -250,7 +250,7 @@
List symbols = new ArrayList();
symbols.add(bigDecimal);
- TupleSource dataSource = new FakeTupleSource(symbols, data);
+ FakeTupleSource dataSource = new FakeTupleSource(symbols, data);
helpProcess(mgr, node, context, expected, dataSource, null);
}
@@ -292,7 +292,7 @@
List symbols = new ArrayList();
symbols.add(col1);
symbols.add(bigDecimal);
- TupleSource dataSource = new FakeTupleSource(symbols, data);
+ FakeTupleSource dataSource = new FakeTupleSource(symbols, data);
helpProcess(mgr, node, context, expected, dataSource, null);
}
@@ -385,7 +385,7 @@
List symbols = new ArrayList();
symbols.add(col1);
symbols.add(bigDecimal);
- TupleSource dataSource = new FakeTupleSource(symbols, data);
+ FakeTupleSource dataSource = new FakeTupleSource(symbols, data);
helpProcess(mgr, node, context, expected, dataSource, null);
}
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 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/test/java/org/teiid/query/processor/relational/TestProjectIntoNode.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -22,42 +22,42 @@
package org.teiid.query.processor.relational;
+import static org.junit.Assert.*;
+
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
+import org.junit.Test;
import org.teiid.api.exception.query.ExpressionEvaluationException;
import org.teiid.common.buffer.BlockedException;
import org.teiid.common.buffer.BufferManager;
import org.teiid.common.buffer.TupleBatch;
import org.teiid.common.buffer.TupleSource;
import org.teiid.core.TeiidComponentException;
+import org.teiid.core.TeiidProcessingException;
+import org.teiid.query.eval.Evaluator;
import org.teiid.query.processor.FakeTupleSource;
import org.teiid.query.processor.ProcessorDataManager;
-import org.teiid.query.processor.relational.ProjectIntoNode;
-import org.teiid.query.processor.relational.RelationalNode;
+import org.teiid.query.processor.relational.ProjectIntoNode.Mode;
import org.teiid.query.sql.lang.BatchedUpdateCommand;
import org.teiid.query.sql.lang.Command;
import org.teiid.query.sql.lang.Insert;
import org.teiid.query.sql.symbol.Constant;
import org.teiid.query.sql.symbol.ElementSymbol;
+import org.teiid.query.sql.symbol.Expression;
import org.teiid.query.sql.symbol.GroupSymbol;
-import org.teiid.query.tempdata.TempTableStore;
import org.teiid.query.util.CommandContext;
-import junit.framework.TestCase;
-
-
-
/**
* @since 4.2
*/
-public class TestProjectIntoNode extends TestCase {
+public class TestProjectIntoNode {
// Rows should be a multiple of batch size for this test to work
private static final int NUM_ROWS = 1000;
- private void helpTestNextBatch(int tupleBatchSize, boolean doBatching, boolean doBulkInsert, boolean exceptionOnClose) throws Exception {
+ private void helpTestNextBatch(int tupleBatchSize, Mode mode) throws Exception {
ProjectIntoNode node = new ProjectIntoNode(2);
@@ -73,14 +73,13 @@
elements.add(elementSymbol_1);
elements.add(elementSymbol_2);
node.setIntoElements(elements);
- node.setDoBatching(doBatching);
- node.setDoBulkInsert(doBulkInsert);
+ node.setMode(mode);
node.setModelName("myModel"); //$NON-NLS-1$
CommandContext context = new CommandContext();
context.setProcessorID("processorID"); //$NON-NLS-1$
BufferManager bm = NodeTestUtil.getTestBufferManager(tupleBatchSize, tupleBatchSize);
- ProcessorDataManager dataManager = new FakePDM(tupleBatchSize, exceptionOnClose);
+ ProcessorDataManager dataManager = new FakePDM(tupleBatchSize);
child.initialize(context, bm, dataManager);
node.initialize(context, bm, dataManager);
@@ -106,32 +105,34 @@
assertEquals(new Integer(NUM_ROWS), columns[0]);
}
- public void testNextBatch() throws Exception {
- helpTestNextBatch(100, true, false, false);
+ @Test public void testNextBatch() throws Exception {
+ helpTestNextBatch(100, Mode.BATCH);
}
- public void testNextBatch_BulkInsert() throws Exception {
- helpTestNextBatch(100, false, true, false);
+ @Test public void testNextBatch_BulkInsert() throws Exception {
+ helpTestNextBatch(100, Mode.BULK);
}
- public void testNextBatch_NoBatching() throws Exception {
- helpTestNextBatch(100, false, false, false);
+ @Test public void testNextBatch_NoBatching() throws Exception {
+ helpTestNextBatch(100, Mode.SINGLE);
}
- public void testNextBatch_Size20Batches() throws Exception {
- helpTestNextBatch(20, true, false, false);
+ @Test public void testNextBatch_Size20Batches() throws Exception {
+ helpTestNextBatch(20, Mode.BATCH);
}
+ @Test public void testNextBatch_Iterator() throws Exception {
+ helpTestNextBatch(100, Mode.ITERATOR);
+ }
+
private static final class FakePDM implements ProcessorDataManager {
private int expectedBatchSize;
private int callCount = 0;
- private boolean exceptionOnClose;
- private FakePDM(int expectedBatchSize, boolean exceptionOnClose) {
+ private FakePDM(int expectedBatchSize) {
this.expectedBatchSize = expectedBatchSize;
- this.exceptionOnClose = exceptionOnClose;
}
public Object lookupCodeValue(CommandContext context,String codeTableName,String returnElementName,String keyElementName,Object keyValue) throws BlockedException,TeiidComponentException {return null;}
- public TupleSource registerRequest(Object processorID,Command command,String modelName,String connectorBindingId, int nodeID) throws TeiidComponentException, ExpressionEvaluationException {
+ public TupleSource registerRequest(Object processorID,Command command,String modelName,String connectorBindingId, int nodeID) throws TeiidComponentException, TeiidProcessingException {
callCount++;
int batchSize = 1;
@@ -140,15 +141,23 @@
if (command instanceof Insert) {
Insert insert = (Insert)command;
if (insert.isBulk()) {
- List batch = TempTableStore.getBulkRows(insert, insert.getVariables());
+ List batch = getBulkRows(insert, insert.getVariables());
batchSize = batch.size();
assertEquals("Unexpected batch on call " + callCount, expectedBatchSize, batchSize); //$NON-NLS-1$
for (int i = 0; i < batchSize; i++) {
ensureValue2((List)batch.get(i), 2, ((callCount-1) * batchSize) + i + 1);
}
+ } else if (insert.getTupleSource() != null) {
+ TupleSource ts = insert.getTupleSource();
+ List tuple = null;
+ int i = 0;
+ while ((tuple = ts.nextTuple()) != null) {
+ ensureValue2(tuple, 2, ++i);
+ }
+ batchSize = i;
} else {
- ensureValue((Insert)command, 2, callCount);
+ ensureValue(insert, 2, callCount);
}
} else if ( command instanceof BatchedUpdateCommand ){
BatchedUpdateCommand bu = (BatchedUpdateCommand)command;
@@ -161,7 +170,6 @@
}
List counts = Arrays.asList(new Object[] { new Integer(batchSize)});
FakeTupleSource fakeTupleSource = new FakeTupleSource(null, new List[] {counts});
- fakeTupleSource.setExceptionOnClose(this.exceptionOnClose);
return fakeTupleSource;
}
@@ -201,9 +209,36 @@
? null
: Arrays.asList(new Object[] {new Integer(currentRow), Integer.toString(currentRow)});
}
- @Override
- public int available() {
- return 0;
- }
}
+
+ public static List<List<Object>> getBulkRows(Insert insert, List<ElementSymbol> elements) throws ExpressionEvaluationException, BlockedException, TeiidComponentException {
+ int bulkRowCount = 1;
+ if (insert.isBulk()) {
+ Constant c = (Constant)insert.getValues().get(0);
+ bulkRowCount = ((List<?>)c.getValue()).size();
+ }
+
+ List<List<Object>> tuples = new ArrayList<List<Object>>(bulkRowCount);
+
+ for (int row = 0; row < bulkRowCount; row++) {
+ List<Object> currentRow = new ArrayList<Object>(insert.getValues().size());
+ for (ElementSymbol symbol : elements) {
+ int index = insert.getVariables().indexOf(symbol);
+ Object value = null;
+ if (index != -1) {
+ if (insert.isBulk()) {
+ Constant multiValue = (Constant)insert.getValues().get(index);
+ value = ((List<?>)multiValue.getValue()).get(row);
+ } else {
+ Expression expr = (Expression)insert.getValues().get(index);
+ value = Evaluator.evaluate(expr);
+ }
+ }
+ currentRow.add(value);
+ }
+ tuples.add(currentRow);
+ }
+ return tuples;
+ }
+
}
Modified: trunk/engine/src/test/java/org/teiid/query/processor/relational/TestSortNode.java
===================================================================
--- trunk/engine/src/test/java/org/teiid/query/processor/relational/TestSortNode.java 2010-07-28 21:11:54 UTC (rev 2390)
+++ trunk/engine/src/test/java/org/teiid/query/processor/relational/TestSortNode.java 2010-07-29 15:18:33 UTC (rev 2391)
@@ -305,7 +305,7 @@
BufferManager bm = BufferManagerFactory.getStandaloneBufferManager();
TupleBuffer tsid = bm.createTupleBuffer(Arrays.asList(es1), "test", TupleSourceType.PROCESSOR); //$NON-NLS-1$
tsid.addTuple(Arrays.asList(1));
- SortUtility su = new SortUtility(tsid.createIndexedTupleSource(), Arrays.asList(es1), Arrays.asList(Boolean.TRUE), Mode.DUP_REMOVE, bm, "test"); //$NON-NLS-1$
+ SortUtility su = new SortUtility(tsid.createIndexedTupleSource(), Arrays.asList(es1), Arrays.asList(Boolean.TRUE), Mode.DUP_REMOVE, bm, "test", tsid.getSchema()); //$NON-NLS-1$
TupleBuffer out = su.sort();
TupleSource ts = out.createIndexedTupleSource();
assertEquals(Arrays.asList(1), ts.nextTuple());
More information about the teiid-commits
mailing list