You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by vo...@apache.org on 2018/08/30 11:53:33 UTC
[11/38] ignite git commit: IGNITE-4191: MVCC and transactional SQL
support. Joint multi-man-years efforts of Semen Boikov, Igor Seliverstov,
Alexander Paschenko, Igor Sapego, Sergey Kalashnikov, Roman Kondakov,
Pavel Kuznetsov, Ivan Pavlukhin, Andrey Mas
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
index 6b7eddf..9b71f37 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
@@ -69,6 +69,12 @@ import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.QueryCursorImpl;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxSelectForUpdateFuture;
+import org.apache.ignite.internal.processors.cache.distributed.near.TxTopologyVersionFuture;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccQueryTracker;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccUtils;
import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
import org.apache.ignite.internal.processors.cache.query.CacheQueryPartitionInfo;
@@ -90,16 +96,20 @@ import org.apache.ignite.internal.processors.query.GridQueryRowCacheCleaner;
import org.apache.ignite.internal.processors.query.GridQueryTypeDescriptor;
import org.apache.ignite.internal.processors.query.GridRunningQueryInfo;
import org.apache.ignite.internal.processors.query.IgniteSQLException;
+import org.apache.ignite.internal.processors.query.NestedTxMode;
import org.apache.ignite.internal.processors.query.QueryField;
import org.apache.ignite.internal.processors.query.QueryIndexDescriptorImpl;
import org.apache.ignite.internal.processors.query.QueryUtils;
import org.apache.ignite.internal.processors.query.SqlClientContext;
+import org.apache.ignite.internal.processors.query.UpdateSourceIterator;
import org.apache.ignite.internal.processors.query.h2.database.H2RowFactory;
import org.apache.ignite.internal.processors.query.h2.database.H2TreeIndex;
import org.apache.ignite.internal.processors.query.h2.database.io.H2ExtrasInnerIO;
import org.apache.ignite.internal.processors.query.h2.database.io.H2ExtrasLeafIO;
import org.apache.ignite.internal.processors.query.h2.database.io.H2InnerIO;
import org.apache.ignite.internal.processors.query.h2.database.io.H2LeafIO;
+import org.apache.ignite.internal.processors.query.h2.database.io.H2MvccInnerIO;
+import org.apache.ignite.internal.processors.query.h2.database.io.H2MvccLeafIO;
import org.apache.ignite.internal.processors.query.h2.ddl.DdlStatementsProcessor;
import org.apache.ignite.internal.processors.query.h2.dml.DmlUtils;
import org.apache.ignite.internal.processors.query.h2.dml.UpdatePlan;
@@ -110,10 +120,13 @@ import org.apache.ignite.internal.processors.query.h2.opt.GridH2QueryContext;
import org.apache.ignite.internal.processors.query.h2.opt.GridH2Row;
import org.apache.ignite.internal.processors.query.h2.opt.GridH2RowDescriptor;
import org.apache.ignite.internal.processors.query.h2.opt.GridH2Table;
+import org.apache.ignite.internal.processors.query.h2.sql.GridSqlAlias;
+import org.apache.ignite.internal.processors.query.h2.sql.GridSqlAst;
import org.apache.ignite.internal.processors.query.h2.sql.GridSqlQuery;
import org.apache.ignite.internal.processors.query.h2.sql.GridSqlQueryParser;
import org.apache.ignite.internal.processors.query.h2.sql.GridSqlQuerySplitter;
import org.apache.ignite.internal.processors.query.h2.sql.GridSqlStatement;
+import org.apache.ignite.internal.processors.query.h2.sql.GridSqlTable;
import org.apache.ignite.internal.processors.query.h2.sys.SqlSystemTableEngine;
import org.apache.ignite.internal.processors.query.h2.sys.view.SqlSystemViewBaselineNodes;
import org.apache.ignite.internal.processors.query.h2.sys.view.SqlSystemViewNodeAttributes;
@@ -121,6 +134,7 @@ import org.apache.ignite.internal.processors.query.h2.sys.view.SqlSystemViewNode
import org.apache.ignite.internal.processors.query.h2.twostep.GridMapQueryExecutor;
import org.apache.ignite.internal.processors.query.h2.twostep.GridReduceQueryExecutor;
import org.apache.ignite.internal.processors.query.h2.twostep.MapQueryLazyWorker;
+import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2QueryRequest;
import org.apache.ignite.internal.processors.query.h2.sys.view.SqlSystemView;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheVisitor;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheVisitorClosure;
@@ -130,14 +144,17 @@ import org.apache.ignite.internal.sql.SqlParseException;
import org.apache.ignite.internal.sql.SqlParser;
import org.apache.ignite.internal.sql.SqlStrictParseException;
import org.apache.ignite.internal.sql.command.SqlAlterTableCommand;
-import org.apache.ignite.internal.sql.command.SqlBulkLoadCommand;
import org.apache.ignite.internal.sql.command.SqlAlterUserCommand;
+import org.apache.ignite.internal.sql.command.SqlBeginTransactionCommand;
+import org.apache.ignite.internal.sql.command.SqlBulkLoadCommand;
import org.apache.ignite.internal.sql.command.SqlCommand;
+import org.apache.ignite.internal.sql.command.SqlCommitTransactionCommand;
import org.apache.ignite.internal.sql.command.SqlCreateIndexCommand;
import org.apache.ignite.internal.sql.command.SqlCreateUserCommand;
import org.apache.ignite.internal.sql.command.SqlDropIndexCommand;
-import org.apache.ignite.internal.sql.command.SqlSetStreamingCommand;
import org.apache.ignite.internal.sql.command.SqlDropUserCommand;
+import org.apache.ignite.internal.sql.command.SqlRollbackTransactionCommand;
+import org.apache.ignite.internal.sql.command.SqlSetStreamingCommand;
import org.apache.ignite.internal.util.GridBoundedConcurrentLinkedHashMap;
import org.apache.ignite.internal.util.GridEmptyCloseableIterator;
import org.apache.ignite.internal.util.GridSpinBusyLock;
@@ -153,6 +170,7 @@ import org.apache.ignite.lang.IgniteBiClosure;
import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.lang.IgniteFuture;
import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.marshaller.Marshaller;
import org.apache.ignite.marshaller.jdk.JdkMarshaller;
import org.apache.ignite.plugin.extensions.communication.Message;
@@ -180,12 +198,17 @@ import static org.apache.ignite.IgniteSystemProperties.IGNITE_H2_INDEXING_CACHE_
import static org.apache.ignite.IgniteSystemProperties.IGNITE_H2_INDEXING_CACHE_THREAD_USAGE_TIMEOUT;
import static org.apache.ignite.IgniteSystemProperties.getInteger;
import static org.apache.ignite.IgniteSystemProperties.getString;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.checkActive;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.mvccEnabled;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.tx;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.txStart;
import static org.apache.ignite.internal.processors.cache.query.GridCacheQueryType.SQL;
import static org.apache.ignite.internal.processors.cache.query.GridCacheQueryType.SQL_FIELDS;
import static org.apache.ignite.internal.processors.cache.query.GridCacheQueryType.TEXT;
import static org.apache.ignite.internal.processors.query.QueryUtils.KEY_FIELD_NAME;
import static org.apache.ignite.internal.processors.query.QueryUtils.VAL_FIELD_NAME;
import static org.apache.ignite.internal.processors.query.QueryUtils.VER_FIELD_NAME;
+import static org.apache.ignite.internal.processors.query.h2.PreparedStatementEx.INVOLVED_CACHES;
import static org.apache.ignite.internal.processors.query.h2.opt.DistributedJoinMode.OFF;
import static org.apache.ignite.internal.processors.query.h2.opt.DistributedJoinMode.distributedJoinMode;
import static org.apache.ignite.internal.processors.query.h2.opt.GridH2QueryType.LOCAL;
@@ -203,13 +226,14 @@ import static org.apache.ignite.internal.processors.query.h2.opt.GridH2QueryType
public class IgniteH2Indexing implements GridQueryIndexing {
/** A pattern for commands having internal implementation in Ignite. */
public static final Pattern INTERNAL_CMD_RE = Pattern.compile(
- "^(create|drop)\\s+index|^alter\\s+table|^copy|^set|^(create|alter|drop)\\s+user", Pattern.CASE_INSENSITIVE);
+ "^(create|drop)\\s+index|^alter\\s+table|^copy|^set|^begin|^commit|^rollback|^(create|alter|drop)\\s+user",
+ Pattern.CASE_INSENSITIVE);
/*
* Register IO for indexes.
*/
static {
- PageIO.registerH2(H2InnerIO.VERSIONS, H2LeafIO.VERSIONS);
+ PageIO.registerH2(H2InnerIO.VERSIONS, H2LeafIO.VERSIONS, H2MvccInnerIO.VERSIONS, H2MvccLeafIO.VERSIONS);
H2ExtrasInnerIO.register();
H2ExtrasLeafIO.register();
@@ -235,19 +259,16 @@ public class IgniteH2Indexing implements GridQueryIndexing {
singletonList(new H2SqlFieldMetadata(null, null, "UPDATED", Long.class.getName(), -1, -1));
/** */
- private static final int PREPARED_STMT_CACHE_SIZE = 256;
-
- /** */
private static final int TWO_STEP_QRY_CACHE_SIZE = 1024;
- /** The period of clean up the {@link #stmtCache}. */
+ /** The period of clean up the statement cache. */
private final Long CLEANUP_STMT_CACHE_PERIOD = Long.getLong(IGNITE_H2_INDEXING_CACHE_CLEANUP_PERIOD, 10_000);
/** The period of clean up the {@link #conns}. */
@SuppressWarnings("FieldCanBeLocal")
private final Long CLEANUP_CONNECTIONS_PERIOD = 2000L;
- /** The timeout to remove entry from the {@link #stmtCache} if the thread doesn't perform any queries. */
+ /** The timeout to remove entry from the statement cache if the thread doesn't perform any queries. */
private final Long STATEMENT_CACHE_THREAD_USAGE_TIMEOUT =
Long.getLong(IGNITE_H2_INDEXING_CACHE_THREAD_USAGE_TIMEOUT, 600 * 1000);
@@ -274,7 +295,8 @@ public class IgniteH2Indexing implements GridQueryIndexing {
private String dbUrl = "jdbc:h2:mem:";
/** */
- private final ConcurrentMap<Thread, Connection> conns = new ConcurrentHashMap<>();
+ // TODO https://issues.apache.org/jira/browse/IGNITE-9062
+ private final ConcurrentMap<Thread, H2ConnectionWrapper> conns = new ConcurrentHashMap<>();
/** */
private GridMapQueryExecutor mapQryExec;
@@ -301,44 +323,39 @@ public class IgniteH2Indexing implements GridQueryIndexing {
private final H2RowCacheRegistry rowCache = new H2RowCacheRegistry();
/** */
- private final ThreadLocal<H2ConnectionWrapper> connCache = new ThreadLocal<H2ConnectionWrapper>() {
- @Override public H2ConnectionWrapper get() {
- H2ConnectionWrapper c = super.get();
+ // TODO https://issues.apache.org/jira/browse/IGNITE-9062
+ private final ThreadLocalObjectPool<H2ConnectionWrapper> connectionPool = new ThreadLocalObjectPool<>(IgniteH2Indexing.this::newConnectionWrapper, 5);
+
+ /** */
+ // TODO https://issues.apache.org/jira/browse/IGNITE-9062
+ private final ThreadLocal<ThreadLocalObjectPool.Reusable<H2ConnectionWrapper>> connCache = new ThreadLocal<ThreadLocalObjectPool.Reusable<H2ConnectionWrapper>>() {
+ @Override public ThreadLocalObjectPool.Reusable<H2ConnectionWrapper> get() {
+ ThreadLocalObjectPool.Reusable<H2ConnectionWrapper> reusable = super.get();
boolean reconnect = true;
try {
- reconnect = c == null || c.connection().isClosed();
+ reconnect = reusable == null || reusable.object().connection().isClosed();
}
catch (SQLException e) {
U.warn(log, "Failed to check connection status.", e);
}
if (reconnect) {
- c = initialValue();
+ reusable = initialValue();
- set(c);
-
- // Reset statement cache when new connection is created.
- stmtCache.remove(Thread.currentThread());
+ set(reusable);
}
- return c;
+ return reusable;
}
- @Override protected H2ConnectionWrapper initialValue() {
- Connection c;
+ @Override protected ThreadLocalObjectPool.Reusable<H2ConnectionWrapper> initialValue() {
+ ThreadLocalObjectPool.Reusable<H2ConnectionWrapper> reusableConnection = connectionPool.borrow();
- try {
- c = DriverManager.getConnection(dbUrl);
- }
- catch (SQLException e) {
- throw new IgniteSQLException("Failed to initialize DB connection: " + dbUrl, e);
- }
+ conns.put(Thread.currentThread(), reusableConnection.object());
- conns.put(Thread.currentThread(), c);
-
- return new H2ConnectionWrapper(c);
+ return reusableConnection;
}
};
@@ -357,9 +374,6 @@ public class IgniteH2Indexing implements GridQueryIndexing {
/** */
private final ConcurrentMap<QueryTable, GridH2Table> dataTables = new ConcurrentHashMap<>();
- /** Statement cache. */
- private final ConcurrentHashMap<Thread, H2StatementCache> stmtCache = new ConcurrentHashMap<>();
-
/** */
private volatile GridBoundedConcurrentLinkedHashMap<H2TwoStepCachedQueryKey, H2TwoStepCachedQuery> twoStepCache =
new GridBoundedConcurrentLinkedHashMap<>(TWO_STEP_QRY_CACHE_SIZE);
@@ -419,6 +433,15 @@ public class IgniteH2Indexing implements GridQueryIndexing {
return sysConn;
}
+ /** */
+ private H2ConnectionWrapper newConnectionWrapper() {
+ try {
+ return new H2ConnectionWrapper(DriverManager.getConnection(dbUrl));
+ } catch (SQLException e) {
+ throw new IgniteSQLException("Failed to initialize DB connection: " + dbUrl, e);
+ }
+ }
+
/**
* @param c Connection.
* @param sql SQL.
@@ -443,7 +466,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
* @throws SQLException If failed.
*/
@SuppressWarnings("ConstantConditions")
- @NotNull private PreparedStatement prepareStatement(Connection c, String sql, boolean useStmtCache)
+ @NotNull public PreparedStatement prepareStatement(Connection c, String sql, boolean useStmtCache)
throws SQLException {
return prepareStatement(c, sql, useStmtCache, false);
}
@@ -468,7 +491,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
PreparedStatement stmt = cache.get(key);
- if (stmt != null && !stmt.isClosed() && !((JdbcStatement)stmt).isCancelled() &&
+ if (stmt != null && !stmt.isClosed() && !stmt.unwrap(JdbcStatement.class).isCancelled() &&
!GridSqlQueryParser.prepared(stmt).needRecompile()) {
assert stmt.getConnection() == c;
@@ -478,9 +501,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
if (cachedOnly)
return null;
- stmt = prepare0(c, sql);
-
- cache.put(key, stmt);
+ cache.put(key, stmt = PreparedStatementExImpl.wrap(prepare0(c, sql)));
return stmt;
}
@@ -503,36 +524,25 @@ public class IgniteH2Indexing implements GridQueryIndexing {
GridH2Table.insertHack(true);
try {
- return c.prepareStatement(sql);
+ return c.prepareStatement(sql, ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY);
}
finally {
GridH2Table.insertHack(false);
}
}
else
- return c.prepareStatement(sql);
+ return c.prepareStatement(sql, ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY);
}
/**
* @return {@link H2StatementCache} associated with current thread.
*/
@NotNull private H2StatementCache getStatementsCacheForCurrentThread() {
- Thread curThread = Thread.currentThread();
-
- H2StatementCache cache = stmtCache.get(curThread);
-
- if (cache == null) {
- H2StatementCache cache0 = new H2StatementCache(PREPARED_STMT_CACHE_SIZE);
-
- cache = stmtCache.putIfAbsent(curThread, cache0);
+ H2StatementCache statementCache = connCache.get().object().statementCache();
- if (cache == null)
- cache = cache0;
- }
-
- cache.updateLastUsage();
+ statementCache.updateLastUsage();
- return cache;
+ return statementCache;
}
/** {@inheritDoc} */
@@ -550,7 +560,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
* @throws IgniteCheckedException In case of error.
*/
private Connection connectionForThread(@Nullable String schema) throws IgniteCheckedException {
- H2ConnectionWrapper c = connCache.get();
+ H2ConnectionWrapper c = connCache.get().object();
if (c == null)
throw new IgniteCheckedException("Failed to get DB connection for thread (check log for details).");
@@ -724,7 +734,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
* Handles SQL exception.
*/
private void onSqlException() {
- Connection conn = connCache.get().connection();
+ Connection conn = connCache.get().object().connection();
connCache.set(null);
@@ -737,8 +747,12 @@ public class IgniteH2Indexing implements GridQueryIndexing {
}
/** {@inheritDoc} */
- @Override public void store(GridCacheContext cctx, GridQueryTypeDescriptor type, CacheDataRow row,
- @Nullable CacheDataRow prevRow, boolean prevRowAvailable) throws IgniteCheckedException {
+ @Override public void store(GridCacheContext cctx,
+ GridQueryTypeDescriptor type,
+ CacheDataRow row,
+ @Nullable CacheDataRow prevRow,
+ boolean prevRowAvailable) throws IgniteCheckedException
+ {
String cacheName = cctx.name();
H2TableDescriptor tbl = tableDescriptor(schema(cacheName), cacheName, type.name());
@@ -746,7 +760,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
if (tbl == null)
return; // Type was rejected.
- tbl.table().update(row, prevRow, prevRowAvailable);
+ tbl.table().update(row, prevRow, prevRowAvailable);
if (tbl.luceneIndex() != null) {
long expireTime = row.expireTime();
@@ -1024,76 +1038,230 @@ public class IgniteH2Indexing implements GridQueryIndexing {
* @param params Query parameters.
* @param filter Cache name and key filter.
* @param enforceJoinOrder Enforce join order of tables in the query.
+ * @param startTx Start transaction flag.
* @param timeout Query timeout in milliseconds.
* @param cancel Query cancel.
* @return Query result.
* @throws IgniteCheckedException If failed.
*/
@SuppressWarnings("unchecked")
- GridQueryFieldsResult queryLocalSqlFields(final String schemaName, final String qry,
+ public GridQueryFieldsResult queryLocalSqlFields(String schemaName, String qry, @Nullable Collection<Object> params,
+ IndexingQueryFilter filter, boolean enforceJoinOrder, boolean startTx, int timeout,
+ GridQueryCancel cancel) throws IgniteCheckedException {
+ return queryLocalSqlFields(schemaName, qry, params, filter, enforceJoinOrder, startTx, timeout, cancel, null);
+ }
+
+ /**
+ * Queries individual fields (generally used by JDBC drivers).
+ *
+ * @param schemaName Schema name.
+ * @param qry Query.
+ * @param params Query parameters.
+ * @param filter Cache name and key filter.
+ * @param enforceJoinOrder Enforce join order of tables in the query.
+ * @param startTx Start transaction flag.
+ * @param timeout Query timeout in milliseconds.
+ * @param cancel Query cancel.
+ * @param mvccTracker Query tracker.
+ * @return Query result.
+ * @throws IgniteCheckedException If failed.
+ */
+ @SuppressWarnings("unchecked")
+ GridQueryFieldsResult queryLocalSqlFields(final String schemaName, String qry,
@Nullable final Collection<Object> params, final IndexingQueryFilter filter, boolean enforceJoinOrder,
- final int timeout, final GridQueryCancel cancel) throws IgniteCheckedException {
- final Connection conn = connectionForSchema(schemaName);
+ boolean startTx, int timeout, final GridQueryCancel cancel,
+ MvccQueryTracker mvccTracker) throws IgniteCheckedException {
- H2Utils.setupConnection(conn, false, enforceJoinOrder);
+ GridNearTxLocal tx = null; boolean mvccEnabled = mvccEnabled(kernalContext());
- final PreparedStatement stmt = preparedStatementWithParams(conn, qry, params, true);
+ assert mvccEnabled || mvccTracker == null;
- if (GridSqlQueryParser.checkMultipleStatements(stmt))
- throw new IgniteSQLException("Multiple statements queries are not supported for local queries");
+ try {
+ final Connection conn = connectionForSchema(schemaName);
- Prepared p = GridSqlQueryParser.prepared(stmt);
+ H2Utils.setupConnection(conn, false, enforceJoinOrder);
- if (DmlStatementsProcessor.isDmlStatement(p)) {
- SqlFieldsQuery fldsQry = new SqlFieldsQuery(qry);
+ PreparedStatement stmt = preparedStatementWithParams(conn, qry, params, true);
- if (params != null)
- fldsQry.setArgs(params.toArray());
+ if (GridSqlQueryParser.checkMultipleStatements(stmt))
+ throw new IgniteSQLException("Multiple statements queries are not supported for local queries");
- fldsQry.setEnforceJoinOrder(enforceJoinOrder);
- fldsQry.setTimeout(timeout, TimeUnit.MILLISECONDS);
+ Prepared p = GridSqlQueryParser.prepared(stmt);
- return dmlProc.updateSqlFieldsLocal(schemaName, conn, p, fldsQry, filter, cancel);
- }
- else if (DdlStatementsProcessor.isDdlStatement(p))
- throw new IgniteSQLException("DDL statements are supported for the whole cluster only",
- IgniteQueryErrorCode.UNSUPPORTED_OPERATION);
+ if (DmlStatementsProcessor.isDmlStatement(p)) {
+ SqlFieldsQuery fldsQry = new SqlFieldsQuery(qry);
- List<GridQueryFieldMetadata> meta;
+ if (params != null)
+ fldsQry.setArgs(params.toArray());
- try {
- meta = H2Utils.meta(stmt.getMetaData());
- }
- catch (SQLException e) {
- throw new IgniteCheckedException("Cannot prepare query metadata", e);
- }
+ fldsQry.setEnforceJoinOrder(enforceJoinOrder);
+ fldsQry.setTimeout(timeout, TimeUnit.MILLISECONDS);
- final GridH2QueryContext ctx = new GridH2QueryContext(nodeId, nodeId, 0, LOCAL)
- .filter(filter).distributedJoinMode(OFF);
+ return dmlProc.updateSqlFieldsLocal(schemaName, conn, p, fldsQry, filter, cancel);
+ }
+ else if (DdlStatementsProcessor.isDdlStatement(p)) {
+ throw new IgniteSQLException("DDL statements are supported for the whole cluster only.",
+ IgniteQueryErrorCode.UNSUPPORTED_OPERATION);
+ }
- return new GridQueryFieldsResultAdapter(meta, null) {
- @Override public GridCloseableIterator<List<?>> iterator() throws IgniteCheckedException {
- assert GridH2QueryContext.get() == null;
+ final GridH2QueryContext ctx = new GridH2QueryContext(nodeId, nodeId, 0, LOCAL)
+ .filter(filter).distributedJoinMode(OFF);
- GridH2QueryContext.set(ctx);
+ boolean forUpdate = GridSqlQueryParser.isForUpdateQuery(p);
- GridRunningQueryInfo run = new GridRunningQueryInfo(qryIdGen.incrementAndGet(), qry, SQL_FIELDS,
- schemaName, U.currentTimeMillis(), cancel, true);
+ if (forUpdate && !mvccEnabled)
+ throw new IgniteSQLException("SELECT FOR UPDATE query requires transactional " +
+ "cache with MVCC enabled.", IgniteQueryErrorCode.UNSUPPORTED_OPERATION);
- runs.putIfAbsent(run.id(), run);
+ GridNearTxSelectForUpdateFuture sfuFut = null;
- try {
- ResultSet rs = executeSqlQueryWithTimer(stmt, conn, qry, params, timeout, cancel);
+ if (mvccEnabled) {
+ if (mvccTracker == null)
+ mvccTracker = mvccTracker(stmt, startTx);
+
+ if (mvccTracker != null) {
+ ctx.mvccSnapshot(mvccTracker.snapshot());
- return new H2FieldsIterator(rs);
+ if ((tx = checkActive(tx(this.ctx))) != null) {
+ int tm1 = (int)tx.remainingTime(), tm2 = timeout;
+
+ timeout = tm1 > 0 && tm2 > 0 ? Math.min(tm1, tm2) : Math.max(tm1, tm2);
+ }
}
- finally {
- GridH2QueryContext.clearThreadLocal();
- runs.remove(run.id());
+ if (forUpdate) {
+ if (mvccTracker == null)
+ throw new IgniteSQLException("SELECT FOR UPDATE query requires transactional " +
+ "cache with MVCC enabled.", IgniteQueryErrorCode.UNSUPPORTED_OPERATION);
+
+ GridSqlStatement stmt0 = new GridSqlQueryParser(false).parse(p);
+
+ qry = GridSqlQueryParser.rewriteQueryForUpdateIfNeeded(stmt0, forUpdate = tx != null);
+
+ stmt = preparedStatementWithParams(conn, qry, params, true);
+
+ if (forUpdate) {
+ GridCacheContext cctx = mvccTracker.context();
+
+ try {
+ if (tx.topologyVersionSnapshot() == null)
+ new TxTopologyVersionFuture(tx, cctx).get();
+ }
+ catch (Exception e) {
+ throw new IgniteSQLException("Failed to lock topology for SELECT FOR UPDATE query.", e);
+ }
+
+ sfuFut = new GridNearTxSelectForUpdateFuture(cctx, tx, timeout);
+
+ sfuFut.initLocal();
+ }
}
}
- };
+
+ List<GridQueryFieldMetadata> meta;
+
+ try {
+ meta = H2Utils.meta(stmt.getMetaData());
+
+ if (forUpdate) {
+ assert meta.size() >= 1;
+
+ meta = meta.subList(0, meta.size() - 1);
+ }
+ }
+ catch (SQLException e) {
+ throw new IgniteCheckedException("Cannot prepare query metadata", e);
+ }
+
+ GridNearTxLocal tx0 = tx;
+ MvccQueryTracker mvccTracker0 = mvccTracker;
+ GridNearTxSelectForUpdateFuture sfuFut0 = sfuFut;
+ PreparedStatement stmt0 = stmt;
+ String qry0 = qry;
+ int timeout0 = timeout;
+
+ return new GridQueryFieldsResultAdapter(meta, null) {
+ @Override public GridCloseableIterator<List<?>> iterator() throws IgniteCheckedException {
+ assert GridH2QueryContext.get() == null;
+
+ GridH2QueryContext.set(ctx);
+
+ GridRunningQueryInfo run = new GridRunningQueryInfo(qryIdGen.incrementAndGet(), qry0,
+ SQL_FIELDS, schemaName, U.currentTimeMillis(), cancel, true);
+
+ runs.putIfAbsent(run.id(), run);
+
+ try {
+ ResultSet rs = executeSqlQueryWithTimer(stmt0, conn, qry0, params, timeout0, cancel);
+
+ if (sfuFut0 != null) {
+ assert tx0.mvccSnapshot() != null;
+
+ ResultSetEnlistFuture enlistFut = ResultSetEnlistFuture.future(
+ IgniteH2Indexing.this.ctx.localNodeId(),
+ tx0.nearXidVersion(),
+ tx0.mvccSnapshot(),
+ tx0.threadId(),
+ IgniteUuid.randomUuid(),
+ -1,
+ null,
+ tx0,
+ timeout0,
+ sfuFut0.cache(),
+ rs
+ );
+
+ enlistFut.listen(new IgniteInClosure<IgniteInternalFuture<Long>>() {
+ @Override public void apply(IgniteInternalFuture<Long> fut) {
+ if (fut.error() != null)
+ sfuFut0.onResult(IgniteH2Indexing.this.ctx.localNodeId(), 0L, false, fut.error());
+ else
+ sfuFut0.onResult(IgniteH2Indexing.this.ctx.localNodeId(), fut.result(), false, null);
+ }
+ });
+
+ enlistFut.init();
+
+ try {
+ sfuFut0.get();
+
+ rs.beforeFirst();
+ }
+ catch (Exception e) {
+ U.closeQuiet(rs);
+
+ throw new IgniteSQLException("Failed to obtain locks on result of SELECT FOR UPDATE.",
+ e);
+ }
+ }
+
+ return new H2FieldsIterator(rs, mvccTracker0, sfuFut0 != null);
+ }
+ catch (IgniteCheckedException | RuntimeException | Error e) {
+ try {
+ if (mvccTracker0 != null)
+ mvccTracker0.onDone();
+ }
+ catch (Exception e0) {
+ e.addSuppressed(e0);
+ }
+
+ throw e;
+ }
+ finally {
+ GridH2QueryContext.clearThreadLocal();
+
+ runs.remove(run.id());
+ }
+ }
+ };
+ }
+ catch (IgniteCheckedException | RuntimeException | Error e) {
+ if (mvccEnabled && (tx != null || (tx = tx(ctx)) != null))
+ tx.setRollbackOnly();
+
+ throw e;
+ }
}
/** {@inheritDoc} */
@@ -1171,7 +1339,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
* @return Prepared statement with set parameters.
* @throws IgniteCheckedException If failed.
*/
- private PreparedStatement preparedStatementWithParams(Connection conn, String sql, Collection<Object> params,
+ public PreparedStatement preparedStatementWithParams(Connection conn, String sql, Collection<Object> params,
boolean useStmtCache) throws IgniteCheckedException {
final PreparedStatement stmt;
@@ -1288,7 +1456,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
* @return Result.
* @throws IgniteCheckedException If failed.
*/
- private ResultSet executeSqlQueryWithTimer(PreparedStatement stmt, Connection conn, String sql,
+ public ResultSet executeSqlQueryWithTimer(PreparedStatement stmt, Connection conn, String sql,
@Nullable Collection<Object> params, int timeoutMillis, @Nullable GridQueryCancel cancel)
throws IgniteCheckedException {
long start = U.currentTimeMillis();
@@ -1344,10 +1512,12 @@ public class IgniteH2Indexing implements GridQueryIndexing {
@Override public FieldsQueryCursor<List<?>> queryLocalSqlFields(String schemaName, SqlFieldsQuery qry,
final boolean keepBinary, IndexingQueryFilter filter, GridQueryCancel cancel) throws IgniteCheckedException {
String sql = qry.getSql();
- Object[] args = qry.getArgs();
+ List<Object> params = F.asList(qry.getArgs());
+ boolean enforceJoinOrder = qry.isEnforceJoinOrder(), startTx = autoStartTx(qry);
+ int timeout = qry.getTimeout();
- final GridQueryFieldsResult res = queryLocalSqlFields(schemaName, sql, F.asList(args), filter,
- qry.isEnforceJoinOrder(), qry.getTimeout(), cancel);
+ final GridQueryFieldsResult res = queryLocalSqlFields(schemaName, sql, params, filter,
+ enforceJoinOrder, startTx, timeout, cancel);
QueryCursorImpl<List<?>> cursor = new QueryCursorImpl<>(new Iterable<List<?>>() {
@SuppressWarnings("NullableProblems")
@@ -1439,8 +1609,17 @@ public class IgniteH2Indexing implements GridQueryIndexing {
H2Utils.setupConnection(conn, false, false);
- GridH2QueryContext.set(new GridH2QueryContext(nodeId, nodeId, 0, LOCAL).filter(filter)
- .distributedJoinMode(OFF));
+ GridH2QueryContext qctx = new GridH2QueryContext(nodeId, nodeId, 0, LOCAL).filter(filter)
+ .distributedJoinMode(OFF);
+
+ PreparedStatement stmt = preparedStatementWithParams(conn, sql, params, true);
+
+ MvccQueryTracker mvccTracker = mvccTracker(stmt, false);
+
+ if (mvccTracker != null)
+ qctx.mvccSnapshot(mvccTracker.snapshot());
+
+ GridH2QueryContext.set(qctx);
GridRunningQueryInfo run = new GridRunningQueryInfo(qryIdGen.incrementAndGet(), qry, SQL, schemaName,
U.currentTimeMillis(), null, true);
@@ -1448,27 +1627,88 @@ public class IgniteH2Indexing implements GridQueryIndexing {
runs.put(run.id(), run);
try {
- ResultSet rs = executeSqlQueryWithTimer(conn, sql, params, true, 0, cancel);
+ ResultSet rs = executeSqlQueryWithTimer(stmt, conn, sql, params, 0, cancel);
return new H2KeyValueIterator(rs);
}
finally {
GridH2QueryContext.clearThreadLocal();
+ if (mvccTracker != null)
+ mvccTracker.onDone();
+
runs.remove(run.id());
}
}
/**
+ * Initialises MVCC filter and returns MVCC query tracker if needed.
+ * @param stmt Prepared statement.
+ * @param startTx Start transaction flag.
+ * @return MVCC query tracker or {@code null} if MVCC is disabled for involved caches.
+ */
+ private MvccQueryTracker mvccTracker(PreparedStatement stmt, boolean startTx) throws IgniteCheckedException {
+ List<GridCacheContext> caches;
+ try {
+ if (stmt.isWrapperFor(PreparedStatementEx.class)) {
+ PreparedStatementEx stmtEx = stmt.unwrap(PreparedStatementEx.class);
+
+ caches = stmtEx.meta(INVOLVED_CACHES);
+
+ if (caches == null)
+ stmtEx.putMeta(INVOLVED_CACHES, caches = parseInvolvedCaches(stmt));
+ }
+ else
+ caches = parseInvolvedCaches(stmt);
+ } catch (SQLException e) {
+ throw new IgniteSQLException(e);
+ }
+
+ GridCacheContext firstCctx = null;
+ boolean mvccEnabled = false;
+
+ for (GridCacheContext cctx : caches) {
+ if (firstCctx == null) {
+ firstCctx = cctx;
+ mvccEnabled = firstCctx.mvccEnabled();
+ }
+ else if (mvccEnabled != cctx.mvccEnabled())
+ throw new IllegalStateException("Using caches with different mvcc settings in same query is forbidden.");
+ }
+
+ return mvccEnabled ? MvccUtils.mvccTracker(firstCctx, startTx) : null;
+ }
+
+ /** */
+ private static List<GridCacheContext> parseInvolvedCaches(PreparedStatement stmt) {
+ GridSqlQueryParser parser = new GridSqlQueryParser(false);
+
+ parser.parse(GridSqlQueryParser.prepared(stmt));
+
+ List<GridCacheContext> involvedCaches = new ArrayList<>();
+
+ for (Object o : parser.objectsMap().values()) {
+ if (o instanceof GridSqlAlias)
+ o = GridSqlAlias.unwrap((GridSqlAst) o);
+ if (o instanceof GridSqlTable && ((GridSqlTable) o).dataTable() != null)
+ involvedCaches.add(((GridSqlTable) o).dataTable().cache());
+ }
+
+ return involvedCaches;
+ }
+
+ /**
* @param schemaName Schema name.
* @param qry Query.
* @param keepCacheObj Flag to keep cache object.
* @param enforceJoinOrder Enforce join order of tables.
+ * @param startTx Start transaction flag.
* @param timeoutMillis Query timeout.
* @param cancel Cancel object.
* @param params Query parameters.
* @param parts Partitions.
* @param lazy Lazy query execution flag.
+ * @param mvccTracker Query tracker.
* @return Iterable result.
*/
private Iterable<List<?>> runQueryTwoStep(
@@ -1476,19 +1716,33 @@ public class IgniteH2Indexing implements GridQueryIndexing {
final GridCacheTwoStepQuery qry,
final boolean keepCacheObj,
final boolean enforceJoinOrder,
+ boolean startTx,
final int timeoutMillis,
final GridQueryCancel cancel,
final Object[] params,
final int[] parts,
- final boolean lazy
- ) {
- return new Iterable<List<?>>() {
- @SuppressWarnings("NullableProblems")
- @Override public Iterator<List<?>> iterator() {
- return rdcQryExec.query(schemaName, qry, keepCacheObj, enforceJoinOrder, timeoutMillis, cancel, params,
- parts, lazy);
- }
- };
+ final boolean lazy,
+ MvccQueryTracker mvccTracker) {
+ assert !qry.mvccEnabled() || !F.isEmpty(qry.cacheIds());
+
+ try {
+ final MvccQueryTracker tracker = mvccTracker == null && qry.mvccEnabled() ?
+ MvccUtils.mvccTracker(ctx.cache().context().cacheContext(qry.cacheIds().get(0)), startTx) : mvccTracker;
+
+ if (qry.forUpdate())
+ qry.forUpdate(checkActive(tx(ctx)) != null);
+
+ return new Iterable<List<?>>() {
+ @SuppressWarnings("NullableProblems")
+ @Override public Iterator<List<?>> iterator() {
+ return rdcQryExec.query(schemaName, qry, keepCacheObj, enforceJoinOrder, timeoutMillis,
+ cancel, params, parts, lazy, tracker);
+ }
+ };
+ }
+ catch (IgniteCheckedException e) {
+ throw new CacheException(e);
+ }
}
/**
@@ -1545,7 +1799,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
fqry.setTimeout(qry.getTimeout(), TimeUnit.MILLISECONDS);
final QueryCursor<List<?>> res =
- querySqlFields(schemaName, fqry, null, keepBinary, true, null).get(0);
+ querySqlFields(schemaName, fqry, null, keepBinary, true, null, null).get(0);
final Iterable<Cache.Entry<K, V>> converted = new Iterable<Cache.Entry<K, V>>() {
@Override public Iterator<Cache.Entry<K, V>> iterator() {
@@ -1581,21 +1835,22 @@ public class IgniteH2Indexing implements GridQueryIndexing {
* Try executing query using native facilities.
*
* @param schemaName Schema name.
- * @param sql Query.
+ * @param qry Query.
* @param cliCtx Client context, or {@code null} if not applicable.
* @return Result or {@code null} if cannot parse/process this query.
*/
- private List<FieldsQueryCursor<List<?>>> tryQueryDistributedSqlFieldsNative(String schemaName, String sql,
+ @SuppressWarnings({"ConstantConditions", "StatementWithEmptyBody"})
+ private List<FieldsQueryCursor<List<?>>> tryQueryDistributedSqlFieldsNative(String schemaName, SqlFieldsQuery qry,
@Nullable SqlClientContext cliCtx) {
// Heuristic check for fast return.
- if (!INTERNAL_CMD_RE.matcher(sql.trim()).find())
+ if (!INTERNAL_CMD_RE.matcher(qry.getSql().trim()).find())
return null;
// Parse.
SqlCommand cmd;
try {
- SqlParser parser = new SqlParser(schemaName, sql);
+ SqlParser parser = new SqlParser(schemaName, qry.getSql());
cmd = parser.nextCommand();
@@ -1603,16 +1858,16 @@ public class IgniteH2Indexing implements GridQueryIndexing {
if (parser.nextCommand() != null)
return null;
- // Currently supported commands are:
- // CREATE/DROP INDEX
- // COPY
- // ALTER TABLE
- // SET STREAMING
- // CREATE/ALTER/DROP USER
- if (!(cmd instanceof SqlCreateIndexCommand || cmd instanceof SqlDropIndexCommand
- || cmd instanceof SqlAlterTableCommand || cmd instanceof SqlBulkLoadCommand
+ if (!(cmd instanceof SqlCreateIndexCommand
+ || cmd instanceof SqlDropIndexCommand
+ || cmd instanceof SqlBeginTransactionCommand
+ || cmd instanceof SqlCommitTransactionCommand
+ || cmd instanceof SqlRollbackTransactionCommand
+ || cmd instanceof SqlBulkLoadCommand
+ || cmd instanceof SqlAlterTableCommand
|| cmd instanceof SqlSetStreamingCommand
- || cmd instanceof SqlCreateUserCommand || cmd instanceof SqlAlterUserCommand
+ || cmd instanceof SqlCreateUserCommand
+ || cmd instanceof SqlAlterUserCommand
|| cmd instanceof SqlDropUserCommand))
return null;
}
@@ -1622,7 +1877,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
catch (Exception e) {
// Cannot parse, return.
if (log.isDebugEnabled())
- log.debug("Failed to parse SQL with native parser [qry=" + sql + ", err=" + e + ']');
+ log.debug("Failed to parse SQL with native parser [qry=" + qry.getSql() + ", err=" + e + ']');
if (!IgniteSystemProperties.getBoolean(IgniteSystemProperties.IGNITE_SQL_PARSER_DISABLE_H2_FALLBACK))
return null;
@@ -1632,36 +1887,40 @@ public class IgniteH2Indexing implements GridQueryIndexing {
if (e instanceof SqlParseException)
code = ((SqlParseException)e).code();
- throw new IgniteSQLException("Failed to parse DDL statement: " + sql + ": " + e.getMessage(),
+ throw new IgniteSQLException("Failed to parse DDL statement: " + qry.getSql() + ": " + e.getMessage(),
code, e);
}
// Execute.
- if (cmd instanceof SqlBulkLoadCommand) {
- FieldsQueryCursor<List<?>> cursor = dmlProc.runNativeDmlStatement(sql, cmd);
-
- return Collections.singletonList(cursor);
- }
- else if (cmd instanceof SqlSetStreamingCommand) {
- if (cliCtx == null)
- throw new IgniteSQLException("SET STREAMING command can only be executed from JDBC or ODBC driver.");
-
- SqlSetStreamingCommand setCmd = (SqlSetStreamingCommand)cmd;
-
- boolean on = setCmd.isTurnOn();
-
- if (on)
- cliCtx.enableStreaming(setCmd.allowOverwrite(), setCmd.flushFrequency(),
- setCmd.perNodeBufferSize(), setCmd.perNodeParallelOperations(), setCmd.isOrdered());
+ try {
+ if (cmd instanceof SqlCreateIndexCommand
+ || cmd instanceof SqlDropIndexCommand
+ || cmd instanceof SqlAlterTableCommand
+ || cmd instanceof SqlCreateUserCommand
+ || cmd instanceof SqlAlterUserCommand
+ || cmd instanceof SqlDropUserCommand)
+ return Collections.singletonList(ddlProc.runDdlStatement(qry.getSql(), cmd));
+ else if (cmd instanceof SqlBulkLoadCommand)
+ return Collections.singletonList(dmlProc.runNativeDmlStatement(qry.getSql(), cmd));
+ else if (cmd instanceof SqlSetStreamingCommand) {
+ if (cliCtx == null)
+ throw new IgniteSQLException("SET STREAMING command can only be executed from JDBC or ODBC driver.");
+
+ SqlSetStreamingCommand setCmd = (SqlSetStreamingCommand)cmd;
+
+ if (setCmd.isTurnOn())
+ cliCtx.enableStreaming(setCmd.allowOverwrite(), setCmd.flushFrequency(),
+ setCmd.perNodeBufferSize(), setCmd.perNodeParallelOperations(), setCmd.isOrdered());
+ else
+ cliCtx.disableStreaming();
+ }
else
- cliCtx.disableStreaming();
+ processTxCommand(cmd, qry);
return Collections.singletonList(H2Utils.zeroCursor());
}
- else {
- FieldsQueryCursor<List<?>> cursor = ddlProc.runDdlStatement(sql, cmd);
-
- return Collections.singletonList(cursor);
+ catch (IgniteCheckedException e) {
+ throw new IgniteSQLException("Failed to execute DDL statement [stmt=" + qry.getSql() + ']', e);
}
}
@@ -1679,88 +1938,207 @@ public class IgniteH2Indexing implements GridQueryIndexing {
IgniteQueryErrorCode.STMT_TYPE_MISMATCH);
}
+ /**
+ * Process transactional command.
+ * @param cmd Command.
+ * @param qry Query.
+ * @throws IgniteCheckedException if failed.
+ */
+ private void processTxCommand(SqlCommand cmd, SqlFieldsQuery qry) throws IgniteCheckedException {
+ if (!mvccEnabled(ctx))
+ throw new IgniteSQLException("MVCC must be enabled in order to invoke transactional operation: " +
+ qry.getSql(), IgniteQueryErrorCode.MVCC_DISABLED);
+
+ NestedTxMode nestedTxMode = qry instanceof SqlFieldsQueryEx ? ((SqlFieldsQueryEx)qry).getNestedTxMode() :
+ NestedTxMode.DEFAULT;
+
+ GridNearTxLocal tx = tx(ctx);
+
+ if (cmd instanceof SqlBeginTransactionCommand) {
+ if (tx != null) {
+ if (nestedTxMode == null)
+ nestedTxMode = NestedTxMode.DEFAULT;
+
+ switch (nestedTxMode) {
+ case COMMIT:
+ doCommit(tx);
+
+ txStart(ctx, qry.getTimeout());
+
+ break;
+
+ case IGNORE:
+ log.warning("Transaction has already been started, ignoring BEGIN command.");
+
+ break;
+
+ case ERROR:
+ throw new IgniteSQLException("Transaction has already been started.",
+ IgniteQueryErrorCode.TRANSACTION_EXISTS);
+
+ default:
+ throw new IgniteSQLException("Unexpected nested transaction handling mode: " +
+ nestedTxMode.name());
+ }
+ }
+ else
+ txStart(ctx, qry.getTimeout());
+ }
+ else if (cmd instanceof SqlCommitTransactionCommand) {
+ // Do nothing if there's no transaction.
+ if (tx != null)
+ doCommit(tx);
+ }
+ else {
+ assert cmd instanceof SqlRollbackTransactionCommand;
+
+ // Do nothing if there's no transaction.
+ if (tx != null)
+ doRollback(tx);
+ }
+ }
+
+ /**
+ * Commit and properly close transaction.
+ * @param tx Transaction.
+ * @throws IgniteCheckedException if failed.
+ */
+ @SuppressWarnings("ThrowFromFinallyBlock")
+ private void doCommit(@NotNull GridNearTxLocal tx) throws IgniteCheckedException {
+ try {
+ if (!tx.isRollbackOnly())
+ tx.commit();
+ }
+ finally {
+ closeTx(tx);
+ }
+ }
+
+ /**
+ * Rollback and properly close transaction.
+ * @param tx Transaction.
+ * @throws IgniteCheckedException if failed.
+ */
+ @SuppressWarnings("ThrowFromFinallyBlock")
+ private void doRollback(@NotNull GridNearTxLocal tx) throws IgniteCheckedException {
+ try {
+ tx.rollback();
+ }
+ finally {
+ closeTx(tx);
+ }
+ }
+
+ /**
+ * Properly close transaction.
+ * @param tx Transaction.
+ * @throws IgniteCheckedException if failed.
+ */
+ private void closeTx(@NotNull GridNearTxLocal tx) throws IgniteCheckedException {
+ try {
+ tx.close();
+ }
+ finally {
+ ctx.cache().context().tm().resetContext();
+ }
+ }
/** {@inheritDoc} */
@SuppressWarnings({"StringEquality", "unchecked"})
@Override public List<FieldsQueryCursor<List<?>>> querySqlFields(String schemaName, SqlFieldsQuery qry,
- @Nullable SqlClientContext cliCtx, boolean keepBinary, boolean failOnMultipleStmts, GridQueryCancel cancel) {
- List<FieldsQueryCursor<List<?>>> res = tryQueryDistributedSqlFieldsNative(schemaName, qry.getSql(), cliCtx);
+ @Nullable SqlClientContext cliCtx, boolean keepBinary, boolean failOnMultipleStmts, MvccQueryTracker tracker,
+ GridQueryCancel cancel) {
+ boolean mvccEnabled = mvccEnabled(ctx), startTx = autoStartTx(qry);
- if (res != null)
- return res;
+ try {
+ List<FieldsQueryCursor<List<?>>> res = tryQueryDistributedSqlFieldsNative(schemaName, qry, cliCtx);
- {
- // First, let's check if we already have a two-step query for this statement...
- H2TwoStepCachedQueryKey cachedQryKey = new H2TwoStepCachedQueryKey(schemaName, qry.getSql(),
- qry.isCollocated(), qry.isDistributedJoins(), qry.isEnforceJoinOrder(), qry.isLocal());
+ if (res != null)
+ return res;
- H2TwoStepCachedQuery cachedQry;
+ {
+ // First, let's check if we already have a two-step query for this statement...
+ H2TwoStepCachedQueryKey cachedQryKey = new H2TwoStepCachedQueryKey(schemaName, qry.getSql(),
+ qry.isCollocated(), qry.isDistributedJoins(), qry.isEnforceJoinOrder(), qry.isLocal());
- if ((cachedQry = twoStepCache.get(cachedQryKey)) != null) {
- checkQueryType(qry, true);
+ H2TwoStepCachedQuery cachedQry;
- GridCacheTwoStepQuery twoStepQry = cachedQry.query().copy();
+ if ((cachedQry = twoStepCache.get(cachedQryKey)) != null) {
+ checkQueryType(qry, true);
- List<GridQueryFieldMetadata> meta = cachedQry.meta();
+ GridCacheTwoStepQuery twoStepQry = cachedQry.query().copy();
- res = Collections.singletonList(doRunDistributedQuery(schemaName, qry, twoStepQry, meta, keepBinary,
- cancel));
+ List<GridQueryFieldMetadata> meta = cachedQry.meta();
- if (!twoStepQry.explain())
- twoStepCache.putIfAbsent(cachedQryKey, new H2TwoStepCachedQuery(meta, twoStepQry.copy()));
+ res = Collections.singletonList(doRunDistributedQuery(schemaName, qry, twoStepQry, meta, keepBinary,
+ startTx, tracker, cancel));
- return res;
+ if (!twoStepQry.explain())
+ twoStepCache.putIfAbsent(cachedQryKey, new H2TwoStepCachedQuery(meta, twoStepQry.copy()));
+
+ return res;
+ }
}
- }
- {
- // Second, let's check if we already have a parsed statement...
- PreparedStatement cachedStmt;
+ {
+ // Second, let's check if we already have a parsed statement...
+ PreparedStatement cachedStmt;
- if ((cachedStmt = cachedStatement(connectionForSchema(schemaName), qry.getSql())) != null) {
- Prepared prepared = GridSqlQueryParser.prepared(cachedStmt);
+ if ((cachedStmt = cachedStatement(connectionForSchema(schemaName), qry.getSql())) != null) {
+ Prepared prepared = GridSqlQueryParser.prepared(cachedStmt);
- // We may use this cached statement only for local queries and non queries.
- if (qry.isLocal() || !prepared.isQuery())
- return (List<FieldsQueryCursor<List<?>>>)doRunPrepared(schemaName, prepared, qry, null,
- null, keepBinary, cancel);
+ // We may use this cached statement only for local queries and non queries.
+ if (qry.isLocal() || !prepared.isQuery())
+ return (List<FieldsQueryCursor<List<?>>>)doRunPrepared(schemaName, prepared, qry, null, null,
+ keepBinary, startTx, tracker, cancel);
+ }
}
- }
- res = new ArrayList<>(1);
+ res = new ArrayList<>(1);
- int firstArg = 0;
+ int firstArg = 0;
- String remainingSql = qry.getSql();
+ String remainingSql = qry.getSql();
- while (remainingSql != null) {
- ParsingResult parseRes = parseAndSplit(schemaName,
- remainingSql != qry.getSql() ? cloneFieldsQuery(qry).setSql(remainingSql) : qry, firstArg);
+ while (remainingSql != null) {
+ ParsingResult parseRes = parseAndSplit(schemaName,
+ remainingSql != qry.getSql() ? cloneFieldsQuery(qry).setSql(remainingSql) : qry, firstArg);
- // Let's avoid second reflection getter call by returning Prepared object too
- Prepared prepared = parseRes.prepared();
+ // Let's avoid second reflection getter call by returning Prepared object too
+ Prepared prepared = parseRes.prepared();
- GridCacheTwoStepQuery twoStepQry = parseRes.twoStepQuery();
+ GridCacheTwoStepQuery twoStepQry = parseRes.twoStepQuery();
- List<GridQueryFieldMetadata> meta = parseRes.meta();
+ List<GridQueryFieldMetadata> meta = parseRes.meta();
- SqlFieldsQuery newQry = parseRes.newQuery();
+ SqlFieldsQuery newQry = parseRes.newQuery();
- remainingSql = parseRes.remainingSql();
+ remainingSql = parseRes.remainingSql();
- if (remainingSql != null && failOnMultipleStmts)
- throw new IgniteSQLException("Multiple statements queries are not supported");
+ if (remainingSql != null && failOnMultipleStmts)
+ throw new IgniteSQLException("Multiple statements queries are not supported");
- firstArg += prepared.getParameters().size();
+ firstArg += prepared.getParameters().size();
- res.addAll(doRunPrepared(schemaName, prepared, newQry, twoStepQry, meta, keepBinary, cancel));
+ res.addAll(doRunPrepared(schemaName, prepared, newQry, twoStepQry, meta, keepBinary, startTx, tracker,
+ cancel));
- if (parseRes.twoStepQuery() != null && parseRes.twoStepQueryKey() != null &&
+ if (parseRes.twoStepQuery() != null && parseRes.twoStepQueryKey() != null &&
!parseRes.twoStepQuery().explain())
- twoStepCache.putIfAbsent(parseRes.twoStepQueryKey(), new H2TwoStepCachedQuery(meta, twoStepQry.copy()));
+ twoStepCache.putIfAbsent(parseRes.twoStepQueryKey(), new H2TwoStepCachedQuery(meta,
+ twoStepQry.copy()));
+ }
+
+ return res;
}
+ catch (RuntimeException | Error e) {
+ GridNearTxLocal tx;
- return res;
+ if (mvccEnabled && (tx = tx(ctx)) != null)
+ tx.setRollbackOnly();
+
+ throw e;
+ }
}
/**
@@ -1771,12 +2149,15 @@ public class IgniteH2Indexing implements GridQueryIndexing {
* @param twoStepQry Two-step query if this query must be executed in a distributed way.
* @param meta Metadata for {@code twoStepQry}.
* @param keepBinary Whether binary objects must not be deserialized automatically.
- * @param cancel Query cancel state holder. @return Query result.
+ * @param startTx Start transactionq flag.
+ * @param tracker MVCC tracker.
+ * @param cancel Query cancel state holder.
+ * @return Query result.
*/
@SuppressWarnings("unchecked")
private List<? extends FieldsQueryCursor<List<?>>> doRunPrepared(String schemaName, Prepared prepared,
- SqlFieldsQuery qry, GridCacheTwoStepQuery twoStepQry,
- List<GridQueryFieldMetadata> meta, boolean keepBinary, GridQueryCancel cancel) {
+ SqlFieldsQuery qry, GridCacheTwoStepQuery twoStepQry, List<GridQueryFieldMetadata> meta, boolean keepBinary,
+ boolean startTx, MvccQueryTracker tracker, GridQueryCancel cancel) {
String sqlQry = qry.getSql();
boolean loc = qry.isLocal();
@@ -1842,7 +2223,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
checkQueryType(qry, true);
return Collections.singletonList(doRunDistributedQuery(schemaName, qry, twoStepQry, meta, keepBinary,
- cancel));
+ startTx, tracker, cancel));
}
// We've encountered a local query, let's just run it.
@@ -1879,7 +2260,8 @@ public class IgniteH2Indexing implements GridQueryIndexing {
PreparedStatement stmt = prepareStatementAndCaches(c, qry.getSql());
if (loc && GridSqlQueryParser.checkMultipleStatements(stmt))
- throw new IgniteSQLException("Multiple statements queries are not supported for local queries");
+ throw new IgniteSQLException("Multiple statements queries are not supported for local queries.",
+ IgniteQueryErrorCode.UNSUPPORTED_OPERATION);
GridSqlQueryParser.PreparedWithRemaining prep = GridSqlQueryParser.preparedWithRemaining(stmt);
@@ -1980,7 +2362,9 @@ public class IgniteH2Indexing implements GridQueryIndexing {
.distributedJoinMode(distributedJoinMode(qry.isLocal(), qry.isDistributedJoins())));
try {
- return new ParsingResult(prepared, newQry, remainingSql, split(prepared, newQry),
+ GridCacheTwoStepQuery twoStepQry = split(prepared, newQry);
+
+ return new ParsingResult(prepared, newQry, remainingSql, twoStepQry,
cachedQryKey, H2Utils.meta(stmt.getMetaData()));
}
catch (IgniteCheckedException e) {
@@ -2041,18 +2425,77 @@ public class IgniteH2Indexing implements GridQueryIndexing {
}
/**
+ * @param qry Sql fields query.autoStartTx(qry)
+ * @return {@code True} if need to start transaction.
+ */
+ public boolean autoStartTx(SqlFieldsQuery qry) {
+ if (!mvccEnabled(ctx))
+ return false;
+
+ return qry instanceof SqlFieldsQueryEx && !((SqlFieldsQueryEx)qry).isAutoCommit() && tx(ctx) == null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public UpdateSourceIterator<?> prepareDistributedUpdate(GridCacheContext<?, ?> cctx, int[] ids,
+ int[] parts, String schema, String qry, Object[] params, int flags,
+ int pageSize, int timeout, AffinityTopologyVersion topVer,
+ MvccSnapshot mvccSnapshot, GridQueryCancel cancel) throws IgniteCheckedException {
+
+ SqlFieldsQuery fldsQry = new SqlFieldsQuery(qry);
+
+ if (params != null)
+ fldsQry.setArgs(params);
+
+ fldsQry.setEnforceJoinOrder(isFlagSet(flags, GridH2QueryRequest.FLAG_ENFORCE_JOIN_ORDER));
+ fldsQry.setTimeout(timeout, TimeUnit.MILLISECONDS);
+ fldsQry.setPageSize(pageSize);
+ fldsQry.setLocal(true);
+
+ boolean loc = true;
+
+ final boolean replicated = isFlagSet(flags, GridH2QueryRequest.FLAG_REPLICATED);
+
+ GridCacheContext<?, ?> cctx0;
+
+ if (!replicated
+ && !F.isEmpty(ids)
+ && (cctx0 = CU.firstPartitioned(cctx.shared(), ids)) != null
+ && cctx0.config().getQueryParallelism() > 1) {
+ fldsQry.setDistributedJoins(true);
+
+ loc = false;
+ }
+
+ Connection conn = connectionForSchema(schema);
+
+ H2Utils.setupConnection(conn, false, fldsQry.isEnforceJoinOrder());
+
+ PreparedStatement stmt = preparedStatementWithParams(conn, fldsQry.getSql(),
+ F.asList(fldsQry.getArgs()), true);
+
+ return dmlProc.prepareDistributedUpdate(schema, conn, stmt, fldsQry, backupFilter(topVer, parts), cancel, loc,
+ topVer, mvccSnapshot);
+ }
+
+ private boolean isFlagSet(int flags, int flag) {
+ return (flags & flag) == flag;
+ }
+
+ /**
* Run distributed query on detected set of partitions.
* @param schemaName Schema name.
* @param qry Original query.
* @param twoStepQry Two-step query.
* @param meta Metadata to set to cursor.
* @param keepBinary Keep binary flag.
+ * @param startTx Start transaction flag.
+ * @param mvccTracker Query tracker.
* @param cancel Cancel handler.
* @return Cursor representing distributed query result.
*/
private FieldsQueryCursor<List<?>> doRunDistributedQuery(String schemaName, SqlFieldsQuery qry,
GridCacheTwoStepQuery twoStepQry, List<GridQueryFieldMetadata> meta, boolean keepBinary,
- GridQueryCancel cancel) {
+ boolean startTx, MvccQueryTracker mvccTracker, GridQueryCancel cancel) {
if (log.isDebugEnabled())
log.debug("Parsed query: `" + qry.getSql() + "` into two step query: " + twoStepQry);
@@ -2074,8 +2517,8 @@ public class IgniteH2Indexing implements GridQueryIndexing {
}
QueryCursorImpl<List<?>> cursor = new QueryCursorImpl<>(
- runQueryTwoStep(schemaName, twoStepQry, keepBinary, qry.isEnforceJoinOrder(), qry.getTimeout(), cancel,
- qry.getArgs(), partitions, qry.isLazy()), cancel);
+ runQueryTwoStep(schemaName, twoStepQry, keepBinary, qry.isEnforceJoinOrder(), startTx, qry.getTimeout(),
+ cancel, qry.getArgs(), partitions, qry.isLazy(), mvccTracker), cancel);
cursor.fieldsMeta(meta);
@@ -2141,10 +2584,11 @@ public class IgniteH2Indexing implements GridQueryIndexing {
}
/**
- * @param cacheIds Caches identifiers.
+ * @param cacheIds Cache IDs.
+ * @param twoStepQry Query.
* @throws IllegalStateException if segmented indices used with non-segmented indices.
*/
- private void checkCacheIndexSegmentation(Collection<Integer> cacheIds) {
+ private void processCaches(List<Integer> cacheIds, GridCacheTwoStepQuery twoStepQry) {
if (cacheIds.isEmpty())
return; // Nothing to check
@@ -2152,11 +2596,21 @@ public class IgniteH2Indexing implements GridQueryIndexing {
int expectedParallelism = 0;
- for (Integer cacheId : cacheIds) {
+ boolean mvccEnabled = false;
+
+ for (int i = 0; i < cacheIds.size(); i++) {
+ Integer cacheId = cacheIds.get(i);
+
GridCacheContext cctx = sharedCtx.cacheContext(cacheId);
assert cctx != null;
+ if (i == 0)
+ mvccEnabled = cctx.mvccEnabled();
+ else if (cctx.mvccEnabled() != mvccEnabled)
+ throw new IllegalStateException("Using caches with different mvcc settings in same query is " +
+ "forbidden.");
+
if (!cctx.isPartitioned())
continue;
@@ -2167,6 +2621,18 @@ public class IgniteH2Indexing implements GridQueryIndexing {
"forbidden.");
}
}
+
+ twoStepQry.mvccEnabled(mvccEnabled);
+
+ if (twoStepQry.forUpdate()) {
+ if (cacheIds.size() != 1)
+ throw new IgniteSQLException("SELECT FOR UPDATE is supported only for queries " +
+ "that involve single transactional cache.");
+
+ if (!mvccEnabled)
+ throw new IgniteSQLException("SELECT FOR UPDATE query requires transactional cache " +
+ "with MVCC enabled.", IgniteQueryErrorCode.UNSUPPORTED_OPERATION);
+ }
}
/**
@@ -2449,10 +2915,10 @@ public class IgniteH2Indexing implements GridQueryIndexing {
/**
* Gets collection of table for given schema name.
*
- * @param cacheName Schema name.
+ * @param cacheName Cache name.
* @return Collection of table descriptors.
*/
- private Collection<H2TableDescriptor> tables(String cacheName) {
+ Collection<H2TableDescriptor> tables(String cacheName) {
H2Schema s = schemas.get(schema(cacheName));
if (s == null)
@@ -2481,28 +2947,32 @@ public class IgniteH2Indexing implements GridQueryIndexing {
}
/**
- * Called periodically by {@link GridTimeoutProcessor} to clean up the {@link #stmtCache}.
+ * Called periodically by {@link GridTimeoutProcessor} to clean up the statement cache.
*/
private void cleanupStatementCache() {
- long cur = U.currentTimeMillis();
+ long now = U.currentTimeMillis();
- for (Iterator<Map.Entry<Thread, H2StatementCache>> it = stmtCache.entrySet().iterator(); it.hasNext(); ) {
- Map.Entry<Thread, H2StatementCache> entry = it.next();
+ for (Iterator<Map.Entry<Thread, H2ConnectionWrapper>> it = conns.entrySet().iterator(); it.hasNext(); ) {
+ Map.Entry<Thread, H2ConnectionWrapper> entry = it.next();
Thread t = entry.getKey();
- if (t.getState() == Thread.State.TERMINATED
- || cur - entry.getValue().lastUsage() > STATEMENT_CACHE_THREAD_USAGE_TIMEOUT)
+ if (t.getState() == Thread.State.TERMINATED) {
+ U.close(entry.getValue(), log);
+
it.remove();
+ }
+ else if (now - entry.getValue().statementCache().lastUsage() > STATEMENT_CACHE_THREAD_USAGE_TIMEOUT)
+ entry.getValue().clearStatementCache();
}
}
/**
- * Called periodically by {@link GridTimeoutProcessor} to clean up the {@link #stmtCache}.
+ * Called periodically by {@link GridTimeoutProcessor} to clean up the {@link #conns}.
*/
private void cleanupConnections() {
- for (Iterator<Map.Entry<Thread, Connection>> it = conns.entrySet().iterator(); it.hasNext(); ) {
- Map.Entry<Thread, Connection> entry = it.next();
+ for (Iterator<Map.Entry<Thread, H2ConnectionWrapper>> it = conns.entrySet().iterator(); it.hasNext(); ) {
+ Map.Entry<Thread, H2ConnectionWrapper> entry = it.next();
Thread t = entry.getKey();
@@ -2515,6 +2985,24 @@ public class IgniteH2Indexing implements GridQueryIndexing {
}
/**
+ * Removes from cache and returns associated with current thread connection.
+ * @return Connection associated with current thread.
+ */
+ public ThreadLocalObjectPool.Reusable<H2ConnectionWrapper> detach() {
+ Thread key = Thread.currentThread();
+
+ ThreadLocalObjectPool.Reusable<H2ConnectionWrapper> reusableConnection = connCache.get();
+
+ H2ConnectionWrapper connection = conns.remove(key);
+
+ connCache.remove();
+
+ assert reusableConnection.object().connection() == connection.connection();
+
+ return reusableConnection;
+ }
+
+ /**
* Rebuild indexes from hash index.
*
* @param cacheName Cache name.
@@ -2529,7 +3017,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
SchemaIndexCacheVisitor visitor = new SchemaIndexCacheVisitorImpl(cctx);
- visitor.visit(new RebuldIndexFromHashClosure(qryMgr));
+ visitor.visit(new RebuildIndexFromHashClosure(qryMgr, cctx.mvccEnabled()));
for (H2TableDescriptor tblDesc : tables(cacheName))
tblDesc.table().markRebuildFromHashInProgress(false);
@@ -2879,7 +3367,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
mapQryExec.cancelLazyWorkers();
- for (Connection c : conns.values())
+ for (H2ConnectionWrapper c : conns.values())
U.close(c, log);
conns.clear();
@@ -2915,6 +3403,17 @@ public class IgniteH2Indexing implements GridQueryIndexing {
}
/** {@inheritDoc} */
+ @Override public void onClientDisconnect() throws IgniteCheckedException {
+ if (!mvccEnabled(ctx))
+ return;
+
+ GridNearTxLocal tx = tx(ctx);
+
+ if (tx != null)
+ doRollback(tx);
+ }
+
+ /** {@inheritDoc} */
@Override public void registerCache(String cacheName, String schemaName, GridCacheContext<?, ?> cctx)
throws IgniteCheckedException {
rowCache.onCacheRegistered(cctx);
@@ -2978,7 +3477,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
}
}
- stmtCache.clear();
+ conns.values().forEach(H2ConnectionWrapper::clearStatementCache);
for (H2TableDescriptor tbl : rmvTbls) {
for (Index idx : tbl.table().getIndexes())
@@ -3138,14 +3637,14 @@ public class IgniteH2Indexing implements GridQueryIndexing {
@Override public void cancelAllQueries() {
mapQryExec.cancelLazyWorkers();
- for (Connection c : conns.values())
+ for (H2ConnectionWrapper c : conns.values())
U.close(c, log);
}
/**
* @return Per-thread connections.
*/
- public Map<Thread, Connection> perThreadConnections() {
+ public Map<Thread, ?> perThreadConnections() {
return conns;
}
@@ -3182,7 +3681,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
//Prohibit usage indices with different numbers of segments in same query.
List<Integer> cacheIds = new ArrayList<>(caches0);
- checkCacheIndexSegmentation(cacheIds);
+ processCaches(cacheIds, twoStepQry);
return cacheIds;
}
@@ -3208,22 +3707,4 @@ public class IgniteH2Indexing implements GridQueryIndexing {
private interface ClIter<X> extends AutoCloseable, Iterator<X> {
// No-op.
}
-
- /** */
- private static class RebuldIndexFromHashClosure implements SchemaIndexCacheVisitorClosure {
- /** */
- private final GridCacheQueryManager qryMgr;
-
- /**
- * @param qryMgr Query manager.
- */
- RebuldIndexFromHashClosure(GridCacheQueryManager qryMgr) {
- this.qryMgr = qryMgr;
- }
-
- /** {@inheritDoc} */
- @Override public void apply(CacheDataRow row) throws IgniteCheckedException {
- qryMgr.store(row, null, false);
- }
- }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/NearResultSetEnlistFuture.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/NearResultSetEnlistFuture.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/NearResultSetEnlistFuture.java
new file mode 100644
index 0000000..1856430
--- /dev/null
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/NearResultSetEnlistFuture.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.query.h2;
+
+import java.sql.ResultSet;
+import java.util.UUID;
+import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxLocalAdapter;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxQueryResultsEnlistFuture;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.lang.IgniteUuid;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ *
+ */
+public class NearResultSetEnlistFuture extends GridNearTxQueryResultsEnlistFuture implements ResultSetEnlistFuture {
+ /** */
+ private static final long serialVersionUID = 877907044489718378L;
+
+ /**
+ * @param nearNodeId Near node ID.
+ * @param nearLockVer Near lock version.
+ * @param mvccSnapshot Mvcc snapshot.
+ * @param threadId Thread ID.
+ * @param nearFutId Near future id.
+ * @param nearMiniId Near mini future id.
+ * @param parts Partitions.
+ * @param tx Transaction.
+ * @param timeout Lock acquisition timeout.
+ * @param cctx Cache context.
+ * @param rs Result set to process.
+ */
+ public NearResultSetEnlistFuture(UUID nearNodeId, GridCacheVersion nearLockVer, MvccSnapshot mvccSnapshot,
+ long threadId, IgniteUuid nearFutId, int nearMiniId, @Nullable int[] parts, GridDhtTxLocalAdapter tx,
+ long timeout, GridCacheContext<?, ?> cctx, ResultSet rs) {
+ super(cctx, (GridNearTxLocal)tx, timeout, ResultSetEnlistFuture.createIterator(rs), 0, true);
+ }
+
+
+ /** {@inheritDoc} */
+ @Override public void onError(Throwable error) {
+ onDone(error);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/PreparedStatementEx.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/PreparedStatementEx.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/PreparedStatementEx.java
new file mode 100644
index 0000000..d254971
--- /dev/null
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/PreparedStatementEx.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.query.h2;
+
+import java.sql.PreparedStatement;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ *
+ */
+public interface PreparedStatementEx extends PreparedStatement {
+ /** */
+ static final AtomicInteger metaIdGenerator = new AtomicInteger();
+
+ /** */
+ static final int INVOLVED_CACHES = metaIdGenerator.getAndIncrement();
+
+ /**
+ * @param id Metadata key.
+ * @return Attached metadata.
+ */
+ @Nullable <T> T meta(int id);
+
+ /**
+ * @param id Metadata key.
+ * @param metaObj Metadata object.
+ */
+ void putMeta(int id, Object metaObj);
+}