You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ni...@apache.org on 2024/09/20 08:29:16 UTC
(ignite) branch txsql updated: IGNITE-22732 Transaction aware SQL (#11438)
This is an automated email from the ASF dual-hosted git repository.
nizhikov pushed a commit to branch txsql
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/txsql by this push:
new 00a0d4041ee IGNITE-22732 Transaction aware SQL (#11438)
00a0d4041ee is described below
commit 00a0d4041eeb1b31a244aee97f7fbbd03cf7f591
Author: Nikolay <ni...@apache.org>
AuthorDate: Fri Sep 20 11:29:09 2024 +0300
IGNITE-22732 Transaction aware SQL (#11438)
---
.idea/inspectionProfiles/Project_Default.xml | 5 +-
.../query/calcite/CalciteQueryProcessor.java | 32 +-
.../query/calcite/exec/AbstractIndexScan.java | 9 +-
.../query/calcite/exec/ExchangeServiceImpl.java | 3 +-
.../query/calcite/exec/ExecutionContext.java | 102 +-
.../query/calcite/exec/ExecutionServiceImpl.java | 12 +-
.../processors/query/calcite/exec/IndexScan.java | 102 +-
.../query/calcite/exec/KeyFilteringCursor.java | 73 ++
.../query/calcite/exec/RuntimeSortedIndex.java | 81 +-
.../query/calcite/exec/SortedListRangeCursor.java | 115 ++
.../processors/query/calcite/exec/TableScan.java | 42 +-
.../calcite/exec/exp/ExpressionFactoryImpl.java | 16 +
.../query/calcite/exec/rel/ModifyNode.java | 46 +-
.../query/calcite/message/MessageType.java | 5 +-
.../query/calcite/message/QueryStartRequest.java | 48 +-
.../query/calcite/message/QueryTxEntry.java | 211 ++++
.../query/calcite/prepare/QueryTemplate.java | 29 +-
.../query/calcite/schema/CacheIndexImpl.java | 100 +-
.../processors/query/calcite/util/Commons.java | 22 +
.../processors/query/calcite/QueryChecker.java | 29 +-
.../query/calcite/exec/KeyFilteringCursorTest.java | 160 +++
.../calcite/exec/LogicalRelImplementorTest.java | 1 +
.../query/calcite/exec/RuntimeSortedIndexTest.java | 1 +
.../calcite/exec/rel/AbstractExecutionTest.java | 3 +-
.../integration/AbstractBasicIntegrationTest.java | 33 +-
.../AbstractBasicIntegrationTransactionalTest.java | 242 ++++
.../integration/AggregatesIntegrationTest.java | 80 +-
.../CalciteBasicSecondaryIndexIntegrationTest.java | 100 +-
.../integration/CorrelatesIntegrationTest.java | 14 +-
.../query/calcite/integration/DataTypesTest.java | 487 ++++----
.../query/calcite/integration/DateTimeTest.java | 20 +-
.../integration/HashSpoolIntegrationTest.java | 16 +-
.../IndexMultiRangeScanIntegrationTest.java | 32 +-
.../integration/IndexScanlIntegrationTest.java | 61 +-
.../calcite/integration/JoinIntegrationTest.java | 27 +-
.../integration/JoinRehashIntegrationTest.java | 6 +-
.../integration/KeepBinaryIntegrationTest.java | 72 +-
.../integration/LimitOffsetIntegrationTest.java | 17 +-
.../integration/LocalDateTimeSupportTest.java | 190 ++--
.../ServerStatisticsIntegrationTest.java | 46 +-
.../calcite/integration/SetOpIntegrationTest.java | 172 ++-
.../integration/SortAggregateIntegrationTest.java | 16 +-
.../integration/TableDmlIntegrationTest.java | 124 +-
...finedFunctionsIntegrationTransactionalTest.java | 46 +
.../query/calcite/planner/PlannerTest.java | 3 +-
.../tx/SqlTransactionsIsolationTest.java | 1196 ++++++++++++++++++++
.../tx/SqlTransactionsUnsupportedModesTest.java | 96 ++
.../ignite/testsuites/IgniteCalciteTestSuite.java | 7 +
.../ignite/testsuites/IntegrationTestSuite.java | 2 +
.../configuration/TransactionConfiguration.java | 33 +
.../org/apache/ignite/internal/IgniteKernal.java | 2 +
.../ignite/internal/IgniteNodeAttributes.java | 3 +
.../query/index/sorted/inline/InlineIndexImpl.java | 87 +-
.../query/index/sorted/inline/InlineIndexTree.java | 21 +-
.../sorted/inline/SortedSegmentedIndexCursor.java | 108 ++
.../cache/ValidationOnNodeJoinUtils.java | 13 +
.../cache/persistence/tree/BPlusTree.java | 10 +
.../cache/transactions/IgniteTxManager.java | 2 +-
.../processors/query/GridQueryProcessor.java | 74 +-
.../apache/ignite/internal/util/lang/GridFunc.java | 16 +
.../H2TransactionAwareQueriesEnabledTest.java | 80 ++
.../IgniteBinaryCacheQueryTestSuite3.java | 5 +-
62 files changed, 3870 insertions(+), 936 deletions(-)
diff --git a/.idea/inspectionProfiles/Project_Default.xml b/.idea/inspectionProfiles/Project_Default.xml
index 0af10ada908..56dbd67f701 100644
--- a/.idea/inspectionProfiles/Project_Default.xml
+++ b/.idea/inspectionProfiles/Project_Default.xml
@@ -347,6 +347,9 @@
<inspection_tool class="AssignmentToDateFieldFromParameter" enabled="true" level="WARNING" enabled_by_default="true">
<option name="ignorePrivateMethods" value="true" />
</inspection_tool>
+ <inspection_tool class="AutoCloseableResource" enabled="true" level="WARNING" enabled_by_default="true">
+ <option name="METHOD_MATCHER_CONFIG" value="java.util.Formatter,format,java.io.Writer,append,com.google.common.base.Preconditions,checkNotNull,org.hibernate.Session,close,java.io.PrintWriter,printf,java.io.PrintStream,printf,org.apache.ignite.IgniteTransactions,txStart" />
+ </inspection_tool>
<inspection_tool class="BadOddness" enabled="true" level="WARNING" enabled_by_default="true" />
<inspection_tool class="BigDecimalEquals" enabled="true" level="WARNING" enabled_by_default="true" />
<inspection_tool class="BooleanMethodIsAlwaysInverted" enabled="false" level="WARNING" enabled_by_default="false" />
@@ -752,7 +755,7 @@
<inspection_tool class="ZeroIndexToHead" enabled="false" level="WARNING" enabled_by_default="false" />
<inspection_tool class="ZeroLengthArrayInitialization" enabled="true" level="WARNING" enabled_by_default="true" />
<inspection_tool class="ZipWithIndex" enabled="false" level="WARNING" enabled_by_default="false" />
- <inspection_tool class="unused" enabled="true" level="ERROR" enabled_by_default="true" editorAttributes="ERRORS_ATTRIBUTES" checkParameterExcludingHierarchy="false">
+ <inspection_tool class="unused" enabled="true" level="WARNING" enabled_by_default="true" editorAttributes="WARNING_ATTRIBUTES" checkParameterExcludingHierarchy="false">
<option name="LOCAL_VARIABLE" value="true" />
<option name="FIELD" value="true" />
<option name="METHOD" value="false" />
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/CalciteQueryProcessor.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/CalciteQueryProcessor.java
index 7f8ac3b8c9b..a21ba720ced 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/CalciteQueryProcessor.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/CalciteQueryProcessor.java
@@ -58,7 +58,9 @@ import org.apache.ignite.configuration.QueryEngineConfiguration;
import org.apache.ignite.events.SqlQueryExecutionEvent;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.processors.GridProcessorAdapter;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
import org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.failure.FailureProcessor;
import org.apache.ignite.internal.processors.query.GridQueryFieldMetadata;
import org.apache.ignite.internal.processors.query.IgniteSQLException;
@@ -120,6 +122,7 @@ import org.apache.ignite.internal.util.typedef.F;
import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.IgniteSystemProperties.getLong;
+import static org.apache.ignite.configuration.TransactionConfiguration.TX_AWARE_QUERIES_SUPPORTED_MODES;
import static org.apache.ignite.events.EventType.EVT_SQL_QUERY_EXECUTION;
/** */
@@ -486,6 +489,8 @@ public class CalciteQueryProcessor extends GridProcessorAdapter implements Query
String sql,
Object... params
) throws IgniteSQLException {
+ ensureTransactionModeSupported(qryCtx);
+
SchemaPlus schema = schemaHolder.schema(schemaName);
assert schema != null : "Schema not found: " + schemaName;
@@ -588,7 +593,32 @@ public class CalciteQueryProcessor extends GridProcessorAdapter implements Query
SqlFieldsQuery sqlFieldsQry = qryCtx.unwrap(SqlFieldsQuery.class);
- return sqlFieldsQry != null ? F.asList(sqlFieldsQry.isLocal(), sqlFieldsQry.isEnforceJoinOrder()) : null;
+ return sqlFieldsQry != null
+ ? F.asList(sqlFieldsQry.isLocal(), sqlFieldsQry.isEnforceJoinOrder(), queryTransactionVersion(qryCtx) == null)
+ : null;
+ }
+
+ /** */
+ private static GridCacheVersion queryTransactionVersion(@Nullable QueryContext qryCtx) {
+ return qryCtx == null ? null : qryCtx.unwrap(GridCacheVersion.class);
+ }
+
+ /** */
+ private void ensureTransactionModeSupported(@Nullable QueryContext qryCtx) {
+ if (!ctx.config().getTransactionConfiguration().isTxAwareQueriesEnabled())
+ return;
+
+ GridCacheVersion ver = queryTransactionVersion(qryCtx);
+
+ if (ver == null)
+ return;
+
+ final GridNearTxLocal userTx = ctx.cache().context().tm().tx(ver);
+
+ if (TX_AWARE_QUERIES_SUPPORTED_MODES.contains(userTx.isolation()))
+ return;
+
+ throw new IllegalStateException("Transaction isolation mode not supported for SQL queries: " + userTx.isolation());
}
/** */
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/AbstractIndexScan.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/AbstractIndexScan.java
index 6e87a7b3e6f..28b40966d94 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/AbstractIndexScan.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/AbstractIndexScan.java
@@ -66,14 +66,14 @@ public abstract class AbstractIndexScan<Row, IdxRow> implements Iterable<Row>, A
/** {@inheritDoc} */
@Override public synchronized Iterator<Row> iterator() {
if (ranges == null)
- return new IteratorImpl(idx.find(null, null, true, true, indexQueryContext()));
+ return new IteratorImpl(indexCursor(null, null, true, true));
IgniteClosure<RangeCondition<Row>, IteratorImpl> clo = range -> {
IdxRow lower = range.lower() == null ? null : row2indexRow(range.lower());
IdxRow upper = range.upper() == null ? null : row2indexRow(range.upper());
return new IteratorImpl(
- idx.find(lower, upper, range.lowerInclude(), range.upperInclude(), indexQueryContext()));
+ indexCursor(lower, upper, range.lowerInclude(), range.upperInclude()));
};
if (!ranges.multiBounds()) {
@@ -88,6 +88,11 @@ public abstract class AbstractIndexScan<Row, IdxRow> implements Iterable<Row>, A
return F.flat(F.iterator(ranges, clo, true));
}
+ /** */
+ protected GridCursor<IdxRow> indexCursor(IdxRow lower, IdxRow upper, boolean lowerInclude, boolean upperInclude) {
+ return idx.find(lower, upper, lowerInclude, upperInclude, indexQueryContext());
+ }
+
/** */
protected abstract IdxRow row2indexRow(Row bound);
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExchangeServiceImpl.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExchangeServiceImpl.java
index b10b69f2213..616ef714309 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExchangeServiceImpl.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExchangeServiceImpl.java
@@ -361,6 +361,7 @@ public class ExchangeServiceImpl extends AbstractService implements ExchangeServ
NoOpMemoryTracker.INSTANCE,
NoOpIoTracker.INSTANCE,
0,
- ImmutableMap.of());
+ ImmutableMap.of(),
+ null);
}
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionContext.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionContext.java
index 8627eb8e37c..2c8068e0316 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionContext.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionContext.java
@@ -18,24 +18,37 @@
package org.apache.ignite.internal.processors.query.calcite.exec;
import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Set;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
+import java.util.function.Function;
import org.apache.calcite.DataContext;
import org.apache.calcite.linq4j.QueryProvider;
import org.apache.calcite.schema.SchemaPlus;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.CacheObject;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
+import org.apache.ignite.internal.processors.cache.persistence.CacheDataRowAdapter;
+import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
import org.apache.ignite.internal.processors.query.calcite.exec.exp.ExpressionFactory;
import org.apache.ignite.internal.processors.query.calcite.exec.exp.ExpressionFactoryImpl;
import org.apache.ignite.internal.processors.query.calcite.exec.tracker.ExecutionNodeMemoryTracker;
import org.apache.ignite.internal.processors.query.calcite.exec.tracker.IoTracker;
import org.apache.ignite.internal.processors.query.calcite.exec.tracker.MemoryTracker;
import org.apache.ignite.internal.processors.query.calcite.exec.tracker.RowTracker;
+import org.apache.ignite.internal.processors.query.calcite.message.QueryTxEntry;
import org.apache.ignite.internal.processors.query.calcite.metadata.ColocationGroup;
import org.apache.ignite.internal.processors.query.calcite.metadata.FragmentDescription;
import org.apache.ignite.internal.processors.query.calcite.prepare.AbstractQueryContext;
@@ -45,8 +58,12 @@ import org.apache.ignite.internal.processors.query.calcite.type.IgniteTypeFactor
import org.apache.ignite.internal.processors.query.calcite.util.Commons;
import org.apache.ignite.internal.processors.query.calcite.util.TypeUtils;
import org.apache.ignite.internal.util.lang.RunnableX;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteBiTuple;
import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.internal.processors.query.calcite.util.Commons.checkRange;
@@ -102,6 +119,9 @@ public class ExecutionContext<Row> extends AbstractQueryContext implements DataC
/** */
private final long timeout;
+ /** */
+ private final Collection<QueryTxEntry> qryTxEntries;
+
/** */
private final long startTs;
@@ -127,7 +147,8 @@ public class ExecutionContext<Row> extends AbstractQueryContext implements DataC
MemoryTracker qryMemoryTracker,
IoTracker ioTracker,
long timeout,
- Map<String, Object> params
+ Map<String, Object> params,
+ @Nullable Collection<QueryTxEntry> qryTxEntries
) {
super(qctx);
@@ -142,6 +163,7 @@ public class ExecutionContext<Row> extends AbstractQueryContext implements DataC
this.ioTracker = ioTracker;
this.params = params;
this.timeout = timeout;
+ this.qryTxEntries = qryTxEntries;
startTs = U.currentTimeMillis();
@@ -289,6 +311,84 @@ public class ExecutionContext<Row> extends AbstractQueryContext implements DataC
correlations[id] = value;
}
+ /**
+ * @return Transaction write map.
+ */
+ public Collection<QueryTxEntry> getQryTxEntries() {
+ return qryTxEntries;
+ }
+
+ /** */
+ public static Collection<QueryTxEntry> transactionChanges(
+ Collection<IgniteTxEntry> writeEntries
+ ) {
+ if (F.isEmpty(writeEntries))
+ return null;
+
+ Collection<QueryTxEntry> res = new ArrayList<>();
+
+ for (IgniteTxEntry e : writeEntries) {
+ CacheObject val = e.value();
+
+ if (!F.isEmpty(e.entryProcessors()))
+ val = e.applyEntryProcessors(val);
+
+ res.add(new QueryTxEntry(e.cacheId(), e.key(), val, e.explicitVersion()));
+ }
+
+ return res;
+ }
+
+ /**
+ * @param cacheId Cache id.
+ * @param parts Partitions set.
+ * @param mapper Mapper to specific data type.
+ * @return First, set of object changed in transaction, second, list of transaction data in required format.
+ * @param <R> Required type.
+ */
+ public <R> IgniteBiTuple<Set<KeyCacheObject>, List<R>> transactionChanges(
+ int cacheId,
+ int[] parts,
+ Function<CacheDataRow, R> mapper
+ ) {
+ if (F.isEmpty(qryTxEntries))
+ return F.t(Collections.emptySet(), Collections.emptyList());
+
+ // Expecting parts are sorted or almost sorted and amount of transaction entries are relatively small.
+ if (parts != null && !F.isSorted(parts))
+ Arrays.sort(parts);
+
+ Set<KeyCacheObject> changedKeys = new HashSet<>(qryTxEntries.size());
+ List<R> newAndUpdatedRows = new ArrayList<>(qryTxEntries.size());
+
+ for (QueryTxEntry e : qryTxEntries) {
+ int part = e.key().partition();
+
+ assert part != -1;
+
+ if (e.cacheId() != cacheId)
+ continue;
+
+ if (parts != null && Arrays.binarySearch(parts, part) < 0)
+ continue;
+
+ changedKeys.add(e.key());
+
+ CacheObject val = e.value();
+
+ if (val != null) { // Mix only updated or inserted entries. In case val == null entry removed.
+ newAndUpdatedRows.add(mapper.apply(new CacheDataRowAdapter(
+ e.key(),
+ val,
+ e.version(),
+ CU.EXPIRE_TIME_ETERNAL // Expire time calculated on commit, can use eternal here.
+ )));
+ }
+ }
+
+ return F.t(changedKeys, newAndUpdatedRows);
+ }
+
/**
* Executes a query task.
*
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionServiceImpl.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionServiceImpl.java
index bca13901aaf..252358a870f 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionServiceImpl.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionServiceImpl.java
@@ -47,6 +47,7 @@ import org.apache.ignite.internal.processors.cache.CacheObjectUtils;
import org.apache.ignite.internal.processors.cache.CacheObjectValueContext;
import org.apache.ignite.internal.processors.cache.GridCachePartitionExchangeManager;
import org.apache.ignite.internal.processors.cache.QueryCursorImpl;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
import org.apache.ignite.internal.processors.cache.query.CacheQueryType;
import org.apache.ignite.internal.processors.cache.query.GridCacheQueryType;
import org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode;
@@ -608,6 +609,8 @@ public class ExecutionServiceImpl<Row> extends AbstractService implements Execut
MemoryTracker qryMemoryTracker = qry.createMemoryTracker(memoryTracker, cfg.getQueryMemoryQuota());
+ final GridNearTxLocal userTx = Commons.queryTransaction(qry.context(), ctx.cache().context());
+
ExecutionContext<Row> ectx = new ExecutionContext<>(
qry.context(),
taskExecutor(),
@@ -620,7 +623,8 @@ public class ExecutionServiceImpl<Row> extends AbstractService implements Execut
qryMemoryTracker,
createIoTracker(locNodeId, qry.localQueryId()),
timeout,
- qryParams);
+ qryParams,
+ userTx == null ? null : ExecutionContext.transactionChanges(userTx.writeEntries()));
Node<Row> node = new LogicalRelImplementor<>(ectx, partitionService(), mailboxRegistry(),
exchangeService(), failureProcessor()).go(fragment.root());
@@ -659,7 +663,8 @@ public class ExecutionServiceImpl<Row> extends AbstractService implements Execut
fragmentsPerNode.get(nodeId).intValue(),
qry.parameters(),
parametersMarshalled,
- timeout
+ timeout,
+ ectx.getQryTxEntries()
);
messageService().send(nodeId, req);
@@ -867,7 +872,8 @@ public class ExecutionServiceImpl<Row> extends AbstractService implements Execut
qry.createMemoryTracker(memoryTracker, cfg.getQueryMemoryQuota()),
createIoTracker(nodeId, msg.originatingQryId()),
msg.timeout(),
- Commons.parametersMap(msg.parameters())
+ Commons.parametersMap(msg.parameters()),
+ msg.queryTransactionEntries()
);
executeFragment(qry, (FragmentPlan)qryPlan, ectx);
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/IndexScan.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/IndexScan.java
index d69d7c41334..e30974f50d9 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/IndexScan.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/IndexScan.java
@@ -21,6 +21,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
+import java.util.Set;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.util.ImmutableBitSet;
import org.apache.calcite.util.ImmutableIntList;
@@ -31,15 +32,19 @@ import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.cache.query.index.sorted.IndexKeyType;
import org.apache.ignite.internal.cache.query.index.sorted.IndexPlainRowImpl;
import org.apache.ignite.internal.cache.query.index.sorted.IndexRow;
+import org.apache.ignite.internal.cache.query.index.sorted.IndexRowImpl;
import org.apache.ignite.internal.cache.query.index.sorted.InlineIndexRowHandler;
import org.apache.ignite.internal.cache.query.index.sorted.inline.IndexQueryContext;
import org.apache.ignite.internal.cache.query.index.sorted.inline.InlineIndex;
import org.apache.ignite.internal.cache.query.index.sorted.inline.InlineIndexKeyType;
+import org.apache.ignite.internal.cache.query.index.sorted.inline.InlineIndexTree;
+import org.apache.ignite.internal.cache.query.index.sorted.inline.SortedSegmentedIndexCursor;
import org.apache.ignite.internal.cache.query.index.sorted.inline.io.InlineIO;
import org.apache.ignite.internal.cache.query.index.sorted.keys.IndexKey;
import org.apache.ignite.internal.cache.query.index.sorted.keys.IndexKeyFactory;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState;
@@ -54,6 +59,7 @@ import org.apache.ignite.internal.processors.query.calcite.util.TypeUtils;
import org.apache.ignite.internal.util.lang.GridCursor;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.spi.indexing.IndexingQueryFilter;
import org.apache.ignite.spi.indexing.IndexingQueryFilterImpl;
import org.jetbrains.annotations.Nullable;
@@ -98,6 +104,12 @@ public class IndexScan<Row> extends AbstractIndexScan<Row, IndexRow> {
/** Types of key fields stored in index. */
private final Type[] fieldsStoreTypes;
+ /**
+ * First, set of keys changed (inserted, updated or removed) inside transaction: must be skiped during index scan.
+ * Second, list of rows inserted or updated inside transaction: must be mixed with the scan results.
+ */
+ private final IgniteBiTuple<Set<KeyCacheObject>, List<IndexRow>> txChanges;
+
/**
* @param ectx Execution context.
* @param desc Table descriptor.
@@ -159,6 +171,21 @@ public class IndexScan<Row> extends AbstractIndexScan<Row, IndexRow> {
fieldsStoreTypes[i] = typeFactory.getResultClass(srcRowType.getFieldList().get(i).getType());
fieldIdxMapping = fieldToInlinedKeysMapping(srcRowType.getFieldCount());
+
+ if (!F.isEmpty(ectx.getQryTxEntries())) {
+ InlineIndexRowHandler rowHnd = idx.segment(0).rowHandler();
+
+ txChanges = ectx.transactionChanges(
+ cctx.cacheId(),
+ parts,
+ r -> new IndexRowImpl(rowHnd, r)
+ );
+
+ txChanges.get2().sort(this::compare);
+ }
+ else
+ txChanges = null;
+
}
/**
@@ -221,6 +248,32 @@ public class IndexScan<Row> extends AbstractIndexScan<Row, IndexRow> {
}
}
+ /** {@inheritDoc} */
+ @Override protected GridCursor<IndexRow> indexCursor(IndexRow lower, IndexRow upper, boolean lowerInclude, boolean upperInclude) {
+ GridCursor<IndexRow> idxCursor = super.indexCursor(lower, upper, lowerInclude, upperInclude);
+
+ if (txChanges == null)
+ return idxCursor;
+
+ // `txChanges` returns single thread data structures e.g. `HashSet`, `ArrayList`.
+ // It safe to use them in multiple `FilteredCursor` instances, because, multi range index scan will be flat to the single cursor.
+ // See AbstractIndexScan#iterator.
+ try {
+ return new SortedSegmentedIndexCursor(
+ new GridCursor[]{
+ // This call will change `txChanges.get1()` content.
+ // Removing found key from set more efficient so we break some rules here.
+ new KeyFilteringCursor<>(idxCursor, txChanges.get1(), r -> r.cacheDataRow().key()),
+ new SortedListRangeCursor<>(this::compare, txChanges.get2(), lower, upper, lowerInclude, upperInclude)
+ },
+ idx.indexDefinition()
+ );
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteException(e);
+ }
+ }
+
/** {@inheritDoc} */
@Override protected IndexRow row2indexRow(Row bound) {
if (bound == null)
@@ -362,7 +415,7 @@ public class IndexScan<Row> extends AbstractIndexScan<Row, IndexRow> {
InlineIndexRowHandler rowHnd = idx.segment(0).rowHandler();
- InlineIndexRowFactory rowFactory = isInlineScan() ?
+ InlineIndexRowFactory rowFactory = (isInlineScan() && (txChanges == null || F.isEmpty(txChanges.get1()))) ?
new InlineIndexRowFactory(rowHnd.inlineIndexKeyTypes().toArray(new InlineIndexKeyType[0]), rowHnd) : null;
BPlusTree.TreeRowClosure<IndexRow, IndexRow> rowFilter = isInlineScan() ? null : createNotExpiredRowFilter();
@@ -444,6 +497,8 @@ public class IndexScan<Row> extends AbstractIndexScan<Row, IndexRow> {
InlineIndexKeyType keyType = F.isEmpty(inlineKeyTypes) ? null : inlineKeyTypes.get(0);
return new BPlusTree.TreeRowClosure<IndexRow, IndexRow>() {
+ private IndexRow idxRow;
+
/** {@inheritDoc} */
@Override public boolean apply(
BPlusTree<IndexRow, IndexRow> tree,
@@ -454,11 +509,14 @@ public class IndexScan<Row> extends AbstractIndexScan<Row, IndexRow> {
if (!checkExpired && keyType != null && io instanceof InlineIO) {
Boolean keyIsNull = keyType.isNull(pageAddr, io.offset(idx), ((InlineIO)io).inlineSize());
- if (keyIsNull == Boolean.TRUE)
+ if (keyIsNull == Boolean.TRUE) {
+ idxRow = null;
+
return false;
+ }
}
- IndexRow idxRow = io.getLookupRow(tree, pageAddr, idx);
+ idxRow = io.getLookupRow(tree, pageAddr, idx);
if (checkExpired &&
idxRow.cacheDataRow().expireTime() > 0 &&
@@ -467,17 +525,35 @@ public class IndexScan<Row> extends AbstractIndexScan<Row, IndexRow> {
return idxRow.key(0).type() != IndexKeyType.NULL;
}
+
+ /** {@inheritDoc} */
+ @Override public IndexRow lastRow() {
+ return idxRow;
+ }
};
}
/** */
public static BPlusTree.TreeRowClosure<IndexRow, IndexRow> createNotExpiredRowFilter() {
- return (tree, io, pageAddr, idx) -> {
- IndexRow idxRow = io.getLookupRow(tree, pageAddr, idx);
+ return new BPlusTree.TreeRowClosure<IndexRow, IndexRow>() {
+ private IndexRow idxRow;
+
+ @Override public boolean apply(
+ BPlusTree<IndexRow, IndexRow> tree,
+ BPlusIO<IndexRow> io,
+ long pageAddr,
+ int idx
+ ) throws IgniteCheckedException {
+ idxRow = io.getLookupRow(tree, pageAddr, idx);
- // Skip expired.
- return !(idxRow.cacheDataRow().expireTime() > 0 &&
- idxRow.cacheDataRow().expireTime() <= U.currentTimeMillis());
+ // Skip expired.
+ return !(idxRow.cacheDataRow().expireTime() > 0 &&
+ idxRow.cacheDataRow().expireTime() <= U.currentTimeMillis());
+ }
+
+ @Override public IndexRow lastRow() {
+ return idxRow;
+ }
};
}
@@ -507,4 +583,14 @@ public class IndexScan<Row> extends AbstractIndexScan<Row, IndexRow> {
}
}
}
+
+ /** */
+ private int compare(IndexRow o1, IndexRow o2) {
+ try {
+ return InlineIndexTree.compareFullRows(o1, o2, 0, idx.segment(0).rowHandler(), idx.indexDefinition().rowComparator());
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteException(e);
+ }
+ }
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/KeyFilteringCursor.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/KeyFilteringCursor.java
new file mode 100644
index 00000000000..4e0a0f44c50
--- /dev/null
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/KeyFilteringCursor.java
@@ -0,0 +1,73 @@
+/*
+ * 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.calcite.exec;
+
+import java.util.Set;
+import java.util.function.Function;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.util.lang.GridCursor;
+
+/**
+ * Cursor wrapper that skips all entires that maps to any of {@code skipKeys} key.
+ * <b>Note, for the performance reasons content of {@code skipKeys} will be changed during iteration.</b>
+ */
+class KeyFilteringCursor<R> implements GridCursor<R> {
+ /** Underlying cursor. */
+ private final GridCursor<? extends R> cursor;
+
+ /** Rows that must be skiped on {@link #cursor} iteration. */
+ private final Set<KeyCacheObject> skipKeys;
+
+ /** Mapper from row to {@link KeyCacheObject}. */
+ private final Function<R, KeyCacheObject> toKey;
+
+ /**
+ * @param cursor Sorted cursor.
+ * @param skipKeys Keys to skip. <b>Content will be changed during iteration.</b>
+ * @param toKey Mapper from row to {@link KeyCacheObject}.
+ */
+ KeyFilteringCursor(GridCursor<? extends R> cursor, Set<KeyCacheObject> skipKeys, Function<R, KeyCacheObject> toKey) {
+ this.cursor = cursor;
+ this.skipKeys = skipKeys;
+ this.toKey = toKey;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean next() throws IgniteCheckedException {
+ R cur;
+
+ do {
+ if (!cursor.next())
+ return false;
+
+ cur = cursor.get();
+
+ // Intentionally use of `Set#remove` here.
+ // We want perform as few `toKey` as possible.
+ // So we break some rules here to optimize work with the data provided by the underlying cursor.
+ } while (!skipKeys.isEmpty() && skipKeys.remove(toKey.apply(cur)));
+
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public R get() throws IgniteCheckedException {
+ return cursor.get();
+ }
+}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/RuntimeSortedIndex.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/RuntimeSortedIndex.java
index ecb92c451c9..c53a81e28af 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/RuntimeSortedIndex.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/RuntimeSortedIndex.java
@@ -18,7 +18,6 @@ package org.apache.ignite.internal.processors.query.calcite.exec;
import java.util.ArrayList;
import java.util.Comparator;
-import java.util.List;
import java.util.Objects;
import org.apache.calcite.rel.RelCollation;
import org.apache.calcite.rel.type.RelDataType;
@@ -26,7 +25,6 @@ import org.apache.ignite.internal.cache.query.index.sorted.inline.IndexQueryCont
import org.apache.ignite.internal.processors.query.calcite.exec.exp.RangeIterable;
import org.apache.ignite.internal.util.lang.GridCursor;
import org.apache.ignite.internal.util.typedef.F;
-import org.jetbrains.annotations.Nullable;
/**
* Runtime sorted index.
@@ -90,7 +88,7 @@ public class RuntimeSortedIndex<Row> implements RuntimeIndex<Row>, TreeIndex<Row
Row lowerRow = (lowerBound == null) ? null : lower;
Row upperRow = (upperBound == null) ? null : upper;
- return new Cursor(rows, lowerRow, upperRow, lowerInclude, upperInclude);
+ return new SortedListRangeCursor<>(comp, rows, lowerRow, upperRow, lowerInclude, upperInclude);
}
/**
@@ -104,83 +102,6 @@ public class RuntimeSortedIndex<Row> implements RuntimeIndex<Row>, TreeIndex<Row
return new IndexScan(rowType, this, ranges);
}
- /**
- * Cursor to navigate through a sorted list with duplicates.
- */
- private class Cursor implements GridCursor<Row> {
- /** List of rows. */
- private final List<Row> rows;
-
- /** Upper bound. */
- private final Row upper;
-
- /** Include upper bound. */
- private final boolean includeUpper;
-
- /** Current row. */
- private Row row;
-
- /** Current index of list element. */
- private int idx;
-
- /**
- * @param rows List of rows.
- * @param lower Lower bound.
- * @param upper Upper bound.
- * @param lowerInclude {@code True} for inclusive lower bound.
- * @param upperInclude {@code True} for inclusive upper bound.
- */
- Cursor(List<Row> rows, @Nullable Row lower, @Nullable Row upper, boolean lowerInclude, boolean upperInclude) {
- this.rows = rows;
- this.upper = upper;
- this.includeUpper = upperInclude;
-
- idx = lower == null ? 0 : lowerBound(rows, lower, lowerInclude);
- }
-
- /**
- * Searches the lower bound (skipping duplicates) using a binary search.
- *
- * @param rows List of rows.
- * @param bound Lower bound.
- * @return Lower bound position in the list.
- */
- private int lowerBound(List<Row> rows, Row bound, boolean includeBound) {
- int low = 0, high = rows.size() - 1, idx = -1;
-
- while (low <= high) {
- int mid = (high - low) / 2 + low;
- int compRes = comp.compare(rows.get(mid), bound);
-
- if (compRes > 0)
- high = mid - 1;
- else if (compRes == 0 && includeBound) {
- idx = mid;
- high = mid - 1;
- }
- else
- low = mid + 1;
- }
-
- return idx == -1 ? low : idx;
- }
-
- /** {@inheritDoc} */
- @Override public boolean next() {
- if (idx == rows.size() || (upper != null && comp.compare(upper, rows.get(idx)) < (includeUpper ? 0 : 1)))
- return false;
-
- row = rows.get(idx++);
-
- return true;
- }
-
- /** {@inheritDoc} */
- @Override public Row get() {
- return row;
- }
- }
-
/**
*
*/
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/SortedListRangeCursor.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/SortedListRangeCursor.java
new file mode 100644
index 00000000000..b56738b8edc
--- /dev/null
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/SortedListRangeCursor.java
@@ -0,0 +1,115 @@
+/*
+ * 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.calcite.exec;
+
+import java.util.Comparator;
+import java.util.List;
+import org.apache.ignite.internal.util.lang.GridCursor;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Cursor to navigate through a sorted list with duplicates.
+ */
+public class SortedListRangeCursor<Row> implements GridCursor<Row> {
+ /** */
+ private final Comparator<Row> comp;
+
+ /** List of rows. */
+ private final List<Row> rows;
+
+ /** Upper bound. */
+ private final Row upper;
+
+ /** Include upper bound. */
+ private final boolean includeUpper;
+
+ /** Current row. */
+ private Row row;
+
+ /** Current index of list element. */
+ private int idx;
+
+ /**
+ * @param comp Rows comparator.
+ * @param rows List of rows.
+ * @param lower Lower bound.
+ * @param upper Upper bound.
+ * @param lowerInclude {@code True} for inclusive lower bound.
+ * @param upperInclude {@code True} for inclusive upper bound.
+ */
+ public SortedListRangeCursor(
+ Comparator<Row> comp,
+ List<Row> rows,
+ @Nullable Row lower,
+ @Nullable Row upper,
+ boolean lowerInclude,
+ boolean upperInclude
+ ) {
+ this.comp = comp;
+ this.rows = rows;
+ this.upper = upper;
+ this.includeUpper = upperInclude;
+
+ idx = lower == null ? 0 : lowerBound(rows, lower, lowerInclude);
+ }
+
+ /**
+ * Searches the lower bound (skipping duplicates) using a binary search.
+ *
+ * @param rows List of rows.
+ * @param bound Lower bound.
+ * @return Lower bound position in the list.
+ */
+ private int lowerBound(List<Row> rows, Row bound, boolean includeBound) {
+ int low = 0, high = rows.size() - 1, idx = -1;
+
+ while (low <= high) {
+ int mid = (high - low) / 2 + low;
+ int compRes = comp.compare(rows.get(mid), bound);
+
+ if (compRes > 0)
+ high = mid - 1;
+ else if (compRes == 0 && includeBound) {
+ idx = mid;
+ high = mid - 1;
+ }
+ else
+ low = mid + 1;
+ }
+
+ return idx == -1 ? low : idx;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean next() {
+ // Intentionally use `-comp.compare(rows.get(idx), upper)` here.
+ // `InlineIndexTree#compareFullRows` that used here works correctly only for specific parameter order.
+ // First parameter must be tree row and the second search row. See implementation, for details.
+ if (idx == rows.size() || (upper != null && -comp.compare(rows.get(idx), upper) < (includeUpper ? 0 : 1)))
+ return false;
+
+ row = rows.get(idx++);
+
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Row get() {
+ return row;
+ }
+}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/TableScan.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/TableScan.java
index f70d0823169..2b618cf04d7 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/TableScan.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/TableScan.java
@@ -24,23 +24,28 @@ import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.Queue;
+import java.util.Set;
+import java.util.function.Function;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.util.ImmutableBitSet;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cluster.ClusterTopologyException;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionTopology;
import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
+import org.apache.ignite.internal.processors.cache.persistence.CacheSearchRow;
import org.apache.ignite.internal.processors.query.calcite.exec.RowHandler.RowFactory;
import org.apache.ignite.internal.processors.query.calcite.schema.CacheTableDescriptor;
import org.apache.ignite.internal.util.lang.GridCursor;
import org.apache.ignite.internal.util.lang.GridIteratorAdapter;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteBiTuple;
import org.jetbrains.annotations.Nullable;
/** */
@@ -187,6 +192,15 @@ public class TableScan<Row> implements Iterable<Row>, AutoCloseable {
/** */
private GridCursor<? extends CacheDataRow> cur;
+ /**
+ * First, set of keys changed (inserted, updated or removed) inside transaction: must be skiped during index scan.
+ * Second, list of rows inserted or updated inside transaction: must be mixed with the scan results.
+ */
+ private IgniteBiTuple<Set<KeyCacheObject>, List<CacheDataRow>> txChanges;
+
+ /** */
+ private Iterator<CacheDataRow> txIter = Collections.emptyIterator();
+
/** */
private Row next;
@@ -195,6 +209,16 @@ public class TableScan<Row> implements Iterable<Row>, AutoCloseable {
assert reserved != null;
parts = new ArrayDeque<>(reserved);
+
+ if (!F.isEmpty(ectx.getQryTxEntries())) {
+ txChanges = ectx.transactionChanges(
+ cctx.cacheId(),
+ // All partitions scaned for replication cache.
+ // See TableScan#reserve.
+ cctx.isReplicated() ? null : TableScan.this.parts,
+ Function.identity()
+ );
+ }
}
/** {@inheritDoc} */
@@ -237,11 +261,25 @@ public class TableScan<Row> implements Iterable<Row>, AutoCloseable {
break;
cur = part.dataStore().cursor(cctx.cacheId());
+
+ if (txChanges != null) {
+ // This call will change `txChanges.get1()` content.
+ // Removing found key from set more efficient so we break some rules here.
+ if (!F.isEmpty(txChanges.get1()))
+ cur = new KeyFilteringCursor<>(cur, txChanges.get1(), CacheSearchRow::key);
+
+ txIter = F.iterator0(txChanges.get2(), true, e -> e.key().partition() == part.id());
+ }
}
- if (cur.next()) {
- CacheDataRow row = cur.get();
+ CacheDataRow row;
+
+ if (cur.next())
+ row = cur.get();
+ else
+ row = txIter.hasNext() ? txIter.next() : null;
+ if (row != null) {
if (row.expireTime() > 0 && row.expireTime() <= U.currentTimeMillis())
continue;
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/ExpressionFactoryImpl.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/ExpressionFactoryImpl.java
index 187d2739457..d6fb106667b 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/ExpressionFactoryImpl.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/ExpressionFactoryImpl.java
@@ -62,6 +62,7 @@ import org.apache.calcite.rex.RexShuttle;
import org.apache.calcite.rex.RexUtil;
import org.apache.calcite.sql.type.SqlTypeName;
import org.apache.calcite.sql.validate.SqlConformance;
+import org.apache.ignite.internal.binary.BinaryObjectImpl;
import org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
import org.apache.ignite.internal.processors.query.calcite.exec.RowHandler;
import org.apache.ignite.internal.processors.query.calcite.exec.RowHandler.RowFactory;
@@ -245,11 +246,26 @@ public class ExpressionFactoryImpl<Row> implements ExpressionFactory<Row> {
/** */
@SuppressWarnings("rawtypes")
private static int compare(Object o1, Object o2, int nullComparison) {
+ if (o1 instanceof BinaryObjectImpl)
+ return compareBinary(o1, o2, nullComparison);
+
final Comparable c1 = (Comparable)o1;
final Comparable c2 = (Comparable)o2;
return RelFieldCollation.compare(c1, c2, nullComparison);
}
+ /** */
+ private static int compareBinary(Object o1, Object o2, int nullComparison) {
+ if (o1 == o2)
+ return 0;
+ else if (o1 == null)
+ return nullComparison;
+ else if (o2 == null)
+ return -nullComparison;
+
+ return BinaryObjectImpl.compareForDml(o1, o2);
+ }
+
/** {@inheritDoc} */
@Override public Predicate<Row> predicate(RexNode filter, RelDataType rowType) {
return new PredicateImpl(scalar(filter, rowType));
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/ModifyNode.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/ModifyNode.java
index 4491028d0be..5a9f3dac4ef 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/ModifyNode.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/ModifyNode.java
@@ -30,10 +30,13 @@ import org.apache.calcite.rel.core.TableModify;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.GridCacheProxyImpl;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
import org.apache.ignite.internal.processors.query.IgniteSQLException;
import org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
import org.apache.ignite.internal.processors.query.calcite.schema.CacheTableDescriptor;
import org.apache.ignite.internal.processors.query.calcite.schema.ModifyTuple;
+import org.apache.ignite.internal.processors.query.calcite.util.Commons;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
@@ -195,8 +198,27 @@ public class ModifyNode<Row> extends AbstractNode<Row> implements SingleNode<Row
this.tuples = new ArrayList<>(MODIFY_BATCH_SIZE);
GridCacheContext<Object, Object> cctx = desc.cacheContext();
+ GridCacheProxyImpl<Object, Object> cache = cctx.cache().keepBinary();
+ GridNearTxLocal tx = Commons.queryTransaction(context(), cctx.shared());
+
+ if (tx == null)
+ invokeOutsideTransaction(tuples, cache);
+ else
+ invokeInsideTransaction(tuples, cache, tx);
+ }
+
+ /**
+ * Perform data modification without explicit transaction.
+ * @param tuples Modified data.
+ * @param cache Cache.
+ * @throws IgniteCheckedException If failed.
+ */
+ private void invokeOutsideTransaction(
+ List<ModifyTuple> tuples,
+ GridCacheProxyImpl<Object, Object> cache
+ ) throws IgniteCheckedException {
Map<Object, EntryProcessor<Object, Object, Long>> map = invokeMap(tuples);
- Map<Object, EntryProcessorResult<Long>> res = cctx.cache().keepBinary().invokeAll(map);
+ Map<Object, EntryProcessorResult<Long>> res = cache.invokeAll(map);
long updated = res.values().stream().mapToLong(EntryProcessorResult::get).sum();
@@ -212,6 +234,28 @@ public class ModifyNode<Row> extends AbstractNode<Row> implements SingleNode<Row
updatedRows += updated;
}
+ /**
+ * Performs data modification within user transaction.
+ * @param tuples Modified data.
+ * @param cache Cache.
+ * @param userTx Transaction.
+ * @throws IgniteCheckedException If failed.
+ */
+ private void invokeInsideTransaction(
+ List<ModifyTuple> tuples,
+ GridCacheProxyImpl<Object, Object> cache,
+ GridNearTxLocal userTx
+ ) throws IgniteCheckedException {
+ userTx.resume();
+
+ try {
+ invokeOutsideTransaction(tuples, cache);
+ }
+ finally {
+ userTx.suspend();
+ }
+ }
+
/** */
private IgniteSQLException conflictKeysException(List<Object> conflictKeys) {
if (op == TableModify.Operation.INSERT) {
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/MessageType.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/MessageType.java
index d32775364e9..56b57c385c7 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/MessageType.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/MessageType.java
@@ -57,7 +57,10 @@ public enum MessageType {
COLOCATION_GROUP(351, ColocationGroup::new),
/** */
- FRAGMENT_DESCRIPTION(352, FragmentDescription::new);
+ FRAGMENT_DESCRIPTION(352, FragmentDescription::new),
+
+ /** */
+ QUERY_TX_ENTRY(353, QueryTxEntry::new);
/** */
private final int directType;
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/QueryStartRequest.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/QueryStartRequest.java
index b8d5eb8f8f3..4b63bf02567 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/QueryStartRequest.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/QueryStartRequest.java
@@ -18,13 +18,16 @@
package org.apache.ignite.internal.processors.query.calcite.message;
import java.nio.ByteBuffer;
+import java.util.Collection;
import java.util.UUID;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.GridDirectCollection;
import org.apache.ignite.internal.GridDirectTransient;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.query.calcite.metadata.FragmentDescription;
import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
import org.apache.ignite.plugin.extensions.communication.MessageReader;
import org.apache.ignite.plugin.extensions.communication.MessageWriter;
import org.jetbrains.annotations.Nullable;
@@ -64,6 +67,10 @@ public class QueryStartRequest implements MarshalableMessage, ExecutionContextAw
/** */
private long timeout;
+ /** */
+ @GridDirectCollection(QueryTxEntry.class)
+ private @Nullable Collection<QueryTxEntry> qryTxEntries;
+
/** */
@SuppressWarnings("AssignmentOrReturnOfFieldWithMutableType")
public QueryStartRequest(
@@ -76,7 +83,8 @@ public class QueryStartRequest implements MarshalableMessage, ExecutionContextAw
int totalFragmentsCnt,
Object[] params,
@Nullable byte[] paramsBytes,
- long timeout
+ long timeout,
+ Collection<QueryTxEntry> qryTxEntries
) {
this.qryId = qryId;
this.originatingQryId = originatingQryId;
@@ -88,6 +96,7 @@ public class QueryStartRequest implements MarshalableMessage, ExecutionContextAw
this.params = params;
this.paramsBytes = paramsBytes; // If we already have marshalled params, use it.
this.timeout = timeout;
+ this.qryTxEntries = qryTxEntries;
}
/** */
@@ -168,20 +177,39 @@ public class QueryStartRequest implements MarshalableMessage, ExecutionContextAw
return timeout;
}
+ /**
+ * @return Transaction entries to mixin on query processing.
+ */
+ public @Nullable Collection<QueryTxEntry> queryTransactionEntries() {
+ return qryTxEntries;
+ }
+
/** {@inheritDoc} */
@Override public void prepareMarshal(GridCacheSharedContext<?, ?> ctx) throws IgniteCheckedException {
if (paramsBytes == null && params != null)
paramsBytes = U.marshal(ctx, params);
fragmentDesc.prepareMarshal(ctx);
+
+ if (qryTxEntries != null) {
+ for (QueryTxEntry e : qryTxEntries)
+ e.prepareMarshal(ctx);
+ }
}
/** {@inheritDoc} */
@Override public void prepareUnmarshal(GridCacheSharedContext<?, ?> ctx) throws IgniteCheckedException {
+ ClassLoader ldr = U.resolveClassLoader(ctx.gridConfig());
+
if (params == null && paramsBytes != null)
- params = U.unmarshal(ctx, paramsBytes, U.resolveClassLoader(ctx.gridConfig()));
+ params = U.unmarshal(ctx, paramsBytes, ldr);
fragmentDesc.prepareUnmarshal(ctx);
+
+ if (qryTxEntries != null) {
+ for (QueryTxEntry e : qryTxEntries)
+ e.prepareUnmarshal(ctx, ldr);
+ }
}
/** {@inheritDoc} */
@@ -245,6 +273,12 @@ public class QueryStartRequest implements MarshalableMessage, ExecutionContextAw
writer.incrementState();
case 8:
+ if (!writer.writeCollection("qryTxEntries", qryTxEntries, MessageCollectionItemType.MSG))
+ return false;
+
+ writer.incrementState();
+
+ case 9:
if (!writer.writeAffinityTopologyVersion("ver", ver))
return false;
@@ -328,6 +362,14 @@ public class QueryStartRequest implements MarshalableMessage, ExecutionContextAw
reader.incrementState();
case 8:
+ qryTxEntries = reader.readCollection("qryTxEntries", MessageCollectionItemType.MSG);
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 9:
ver = reader.readAffinityTopologyVersion("ver");
if (!reader.isLastRead())
@@ -347,6 +389,6 @@ public class QueryStartRequest implements MarshalableMessage, ExecutionContextAw
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 9;
+ return 10;
}
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/QueryTxEntry.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/QueryTxEntry.java
new file mode 100644
index 00000000000..6ee247e29f0
--- /dev/null
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/QueryTxEntry.java
@@ -0,0 +1,211 @@
+/*
+ * 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.calcite.message;
+
+import java.nio.ByteBuffer;
+import java.util.Collection;
+import java.util.function.Function;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.configuration.TransactionConfiguration;
+import org.apache.ignite.internal.processors.cache.CacheObject;
+import org.apache.ignite.internal.processors.cache.CacheObjectContext;
+import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
+import org.apache.ignite.plugin.extensions.communication.MessageReader;
+import org.apache.ignite.plugin.extensions.communication.MessageWriter;
+
+/**
+ * Class to pass to remote nodes transaction changes.
+ *
+ * @see TransactionConfiguration#setTxAwareQueriesEnabled(boolean)
+ * @see ExecutionContext#transactionChanges(Collection)
+ * @see ExecutionContext#transactionChanges(int, int[], Function)
+ * @see QueryStartRequest#queryTransactionEntries()
+ */
+public class QueryTxEntry implements CalciteMessage {
+ /** Cache id. */
+ private int cacheId;
+
+ /** Entry key. */
+ private KeyCacheObject key;
+
+ /** Entry value. */
+ private CacheObject val;
+
+ /** Entry version. */
+ private GridCacheVersion ver;
+
+ /**
+ * Empty constructor.
+ */
+ public QueryTxEntry() {
+ // No-op.
+ }
+
+ /**
+ * @param cacheId Cache id.
+ * @param key Key.
+ * @param val Value.
+ * @param ver Version.
+ */
+ public QueryTxEntry(int cacheId, KeyCacheObject key, CacheObject val, GridCacheVersion ver) {
+ this.cacheId = cacheId;
+ this.key = key;
+ this.val = val;
+ this.ver = ver;
+ }
+
+ /** @return Cache id. */
+ public int cacheId() {
+ return cacheId;
+ }
+
+ /** @return Entry key. */
+ public KeyCacheObject key() {
+ return key;
+ }
+
+ /** @return Entry value. */
+ public CacheObject value() {
+ return val;
+ }
+
+ /** @return Entry version. */
+ public GridCacheVersion version() {
+ return ver;
+ }
+
+ /** */
+ public void prepareMarshal(GridCacheSharedContext<?, ?> ctx) throws IgniteCheckedException {
+ CacheObjectContext coctx = ctx.cacheContext(cacheId).cacheObjectContext();
+
+ key.prepareMarshal(coctx);
+
+ if (val != null)
+ val.prepareMarshal(coctx);
+ }
+
+ /** */
+ public void prepareUnmarshal(GridCacheSharedContext<?, ?> ctx, ClassLoader ldr) throws IgniteCheckedException {
+ CacheObjectContext coctx = ctx.cacheContext(cacheId).cacheObjectContext();
+
+ key.finishUnmarshal(coctx, ldr);
+
+ if (val != null)
+ val.finishUnmarshal(coctx, ldr);
+
+ }
+
+ /** {@inheritDoc} */
+ @Override public MessageType type() {
+ return MessageType.QUERY_TX_ENTRY;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
+ writer.setBuffer(buf);
+
+ if (!writer.isHeaderWritten()) {
+ if (!writer.writeHeader(directType(), fieldsCount()))
+ return false;
+
+ writer.onHeaderWritten();
+ }
+
+ switch (writer.state()) {
+ case 0:
+ if (!writer.writeInt("cacheId", cacheId))
+ return false;
+
+ writer.incrementState();
+
+ case 1:
+ if (!writer.writeMessage("key", key))
+ return false;
+
+ writer.incrementState();
+
+ case 2:
+ if (!writer.writeMessage("val", val))
+ return false;
+
+ writer.incrementState();
+
+ case 3:
+ if (!writer.writeMessage("ver", ver))
+ return false;
+
+ writer.incrementState();
+
+ }
+
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) {
+ reader.setBuffer(buf);
+
+ if (!reader.beforeMessageRead())
+ return false;
+
+ switch (reader.state()) {
+ case 0:
+ cacheId = reader.readInt("cacheId");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 1:
+ key = reader.readMessage("key");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 2:
+ val = reader.readMessage("val");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 3:
+ ver = reader.readMessage("ver");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ }
+
+ return reader.afterMessageRead(QueryTxEntry.class);
+ }
+
+ /** {@inheritDoc} */
+ @Override public byte fieldsCount() {
+ return 4;
+ }
+}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/QueryTemplate.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/QueryTemplate.java
index 1958488fcc5..5bfd714ee8b 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/QueryTemplate.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/QueryTemplate.java
@@ -18,21 +18,26 @@
package org.apache.ignite.internal.processors.query.calcite.prepare;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicReference;
-
import com.google.common.collect.ImmutableList;
+import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.metadata.RelMetadataQuery;
import org.apache.ignite.internal.processors.query.IgniteSQLException;
import org.apache.ignite.internal.processors.query.calcite.exec.PartitionExtractor;
import org.apache.ignite.internal.processors.query.calcite.exec.partition.PartitionNode;
import org.apache.ignite.internal.processors.query.calcite.metadata.FragmentMappingException;
import org.apache.ignite.internal.processors.query.calcite.metadata.MappingService;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteExchange;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteReceiver;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRel;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteSender;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTableModify;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions;
import org.apache.ignite.internal.processors.query.calcite.util.Commons;
import org.apache.ignite.internal.util.typedef.F;
import org.jetbrains.annotations.NotNull;
@@ -85,7 +90,27 @@ public class QueryTemplate {
else
ex.addSuppressed(e);
- fragments = replace(fragments, e.fragment(), new FragmentSplitter(e.node()).go(e.fragment()));
+ RelNode cutPoint = e.node();
+
+ if (Commons.queryTransactionVersion(ctx) != null && cutPoint instanceof IgniteTableModify) {
+ IgniteRel input = (IgniteRel)cutPoint.getInput(0);
+
+ // Table modify under transaction should always be executed on initiator node.
+ if (!input.distribution().equals(IgniteDistributions.single())) {
+ cutPoint = ((IgniteRel)cutPoint).clone(cutPoint.getCluster(), Collections.singletonList(
+ new IgniteExchange(
+ input.getCluster(),
+ input.getTraitSet().replace(IgniteDistributions.single()),
+ input,
+ IgniteDistributions.single())));
+
+ fragments = replace(fragments, e.fragment(), new Splitter().go((IgniteRel)cutPoint));
+
+ continue;
+ }
+ }
+
+ fragments = replace(fragments, e.fragment(), new FragmentSplitter(cutPoint).go(e.fragment()));
}
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/schema/CacheIndexImpl.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/schema/CacheIndexImpl.java
index 00e4fa8480c..aeed8641c01 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/schema/CacheIndexImpl.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/schema/CacheIndexImpl.java
@@ -19,7 +19,9 @@ package org.apache.ignite.internal.processors.query.calcite.schema;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.Set;
import java.util.UUID;
+import java.util.function.Function;
import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelOptTable;
import org.apache.calcite.rel.RelCollation;
@@ -35,11 +37,15 @@ import org.apache.ignite.internal.cache.query.index.sorted.IndexKeyDefinition;
import org.apache.ignite.internal.cache.query.index.sorted.IndexKeyType;
import org.apache.ignite.internal.cache.query.index.sorted.IndexKeyTypeSettings;
import org.apache.ignite.internal.cache.query.index.sorted.IndexRow;
+import org.apache.ignite.internal.cache.query.index.sorted.InlineIndexRowHandler;
import org.apache.ignite.internal.cache.query.index.sorted.inline.IndexQueryContext;
import org.apache.ignite.internal.cache.query.index.sorted.inline.InlineIndex;
import org.apache.ignite.internal.cache.query.index.sorted.inline.InlineIndexImpl;
import org.apache.ignite.internal.cache.query.index.sorted.inline.InlineIndexKeyType;
import org.apache.ignite.internal.cache.query.index.sorted.inline.InlineIndexKeyTypeRegistry;
+import org.apache.ignite.internal.cache.query.index.sorted.keys.NullIndexKey;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
import org.apache.ignite.internal.processors.cache.persistence.tree.BPlusTree;
import org.apache.ignite.internal.processors.cache.persistence.tree.io.BPlusIO;
import org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
@@ -51,8 +57,11 @@ import org.apache.ignite.internal.processors.query.calcite.prepare.bounds.Search
import org.apache.ignite.internal.processors.query.calcite.rel.logical.IgniteLogicalIndexScan;
import org.apache.ignite.internal.processors.query.calcite.util.Commons;
import org.apache.ignite.internal.processors.query.calcite.util.RexUtils;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.spi.indexing.IndexingQueryFilter;
import org.apache.ignite.spi.indexing.IndexingQueryFilterImpl;
+import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
/**
@@ -155,16 +164,41 @@ public class CacheIndexImpl implements IgniteIndex {
if (idx == null || !grp.nodeIds().contains(ectx.localNodeId()))
return 0;
+ int[] locParts = grp.partitions(ectx.localNodeId());
+
InlineIndex iidx = idx.unwrap(InlineIndex.class);
+ boolean[] skipCheck = new boolean[] {false};
+
+ BPlusTree.TreeRowClosure<IndexRow, IndexRow> rowFilter = countRowFilter(skipCheck, notNull, iidx);
+
+ long cnt = 0;
+
+ if (!F.isEmpty(ectx.getQryTxEntries())) {
+ IgniteBiTuple<Set<KeyCacheObject>, List<CacheDataRow>> txChanges = ectx.transactionChanges(
+ iidx.indexDefinition().cacheInfo().cacheId(),
+ locParts,
+ Function.identity()
+ );
+
+ if (!txChanges.get1().isEmpty()) {
+ // This call will change `txChanges.get1()` content.
+ // Removing found key from set more efficient so we break some rules here.
+ rowFilter = transactionAwareCountRowFilter(rowFilter, txChanges.get1());
+
+ cnt = countTransactionRows(notNull, iidx, txChanges.get2());
+ }
+ }
+
try {
IndexingQueryFilter filter = new IndexingQueryFilterImpl(tbl.descriptor().cacheContext().kernalContext(),
- ectx.topologyVersion(), grp.partitions(ectx.localNodeId()));
+ ectx.topologyVersion(), locParts);
- long cnt = 0;
+ for (int i = 0; i < iidx.segmentsCount(); ++i) {
+ cnt += iidx.count(i, new IndexQueryContext(filter, rowFilter));
- for (int i = 0; i < iidx.segmentsCount(); ++i)
- cnt += iidx.count(i, new IndexQueryContext(filter, countRowFilter(notNull, iidx)));
+ skipCheck[0] = false;
+ }
return cnt;
}
@@ -174,7 +208,7 @@ public class CacheIndexImpl implements IgniteIndex {
}
/** */
- private @Nullable BPlusTree.TreeRowClosure<IndexRow, IndexRow> countRowFilter(boolean notNull, InlineIndex iidx) {
+ private @Nullable BPlusTree.TreeRowClosure<IndexRow, IndexRow> countRowFilter(boolean[] skipCheck, boolean notNull, InlineIndex iidx) {
boolean checkExpired = !tbl.descriptor().cacheContext().config().isEagerTtl();
if (notNull) {
@@ -183,8 +217,6 @@ public class CacheIndexImpl implements IgniteIndex {
BPlusTree.TreeRowClosure<IndexRow, IndexRow> notNullRowFilter = IndexScan.createNotNullRowFilter(iidx, checkExpired);
return new BPlusTree.TreeRowClosure<>() {
- private boolean skipCheck;
-
@Override public boolean apply(
BPlusTree<IndexRow, IndexRow> tree,
BPlusIO<IndexRow> io,
@@ -195,22 +227,72 @@ public class CacheIndexImpl implements IgniteIndex {
// don't need to check it with notNullRowFilter.
// In case of NULL-LAST collation, all values after first null value will be null,
// don't need to check it too.
- if (skipCheck && !checkExpired)
+ if (skipCheck[0] && !checkExpired)
return nullsFirst;
boolean res = notNullRowFilter.apply(tree, io, pageAddr, idx);
if (res == nullsFirst)
- skipCheck = true;
+ skipCheck[0] = true;
return res;
}
+
+ @Override public IndexRow lastRow() {
+ return (skipCheck[0] && !checkExpired)
+ ? null
+ : notNullRowFilter.lastRow();
+ }
};
}
return checkExpired ? IndexScan.createNotExpiredRowFilter() : null;
}
+ /** */
+ private static @NotNull BPlusTree.TreeRowClosure<IndexRow, IndexRow> transactionAwareCountRowFilter(
+ BPlusTree.TreeRowClosure<IndexRow, IndexRow> rowFilter,
+ Set<KeyCacheObject> skipKeys
+ ) {
+ return new BPlusTree.TreeRowClosure<>() {
+ @Override public boolean apply(
+ BPlusTree<IndexRow, IndexRow> tree,
+ BPlusIO<IndexRow> io,
+ long pageAddr,
+ int idx
+ ) throws IgniteCheckedException {
+ if (rowFilter != null && !rowFilter.apply(tree, io, pageAddr, idx))
+ return false;
+
+ if (skipKeys.isEmpty())
+ return true;
+
+ IndexRow row = rowFilter == null ? null : rowFilter.lastRow();
+
+ if (row == null)
+ row = tree.getRow(io, pageAddr, idx);
+
+ return !skipKeys.remove(row.cacheDataRow().key());
+ }
+ };
+ }
+
+ /** */
+ private static long countTransactionRows(boolean notNull, InlineIndex iidx, List<CacheDataRow> changedRows) {
+ InlineIndexRowHandler rowHnd = iidx.segment(0).rowHandler();
+
+ long cnt = 0;
+
+ for (CacheDataRow txRow : changedRows) {
+ if (rowHnd.indexKey(0, txRow) == NullIndexKey.INSTANCE && notNull)
+ continue;
+
+ cnt++;
+ }
+
+ return cnt;
+ }
+
/** {@inheritDoc} */
@Override public List<SearchBounds> toSearchBounds(
RelOptCluster cluster,
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/util/Commons.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/util/Commons.java
index a88d52c8865..e1c8f8dbfab 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/util/Commons.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/util/Commons.java
@@ -54,7 +54,10 @@ import org.apache.ignite.IgniteLogger;
import org.apache.ignite.internal.GridComponent;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode;
+import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.query.IgniteSQLException;
import org.apache.ignite.internal.processors.query.QueryContext;
import org.apache.ignite.internal.processors.query.calcite.exec.RowHandler;
@@ -455,4 +458,23 @@ public final class Commons {
) {
return new MappingQueryContext(ctx, locNodeId, topVer, qryParams);
}
+
+ /**
+ * @param ctx Query context.
+ * @param cctx Grid cache shared context.
+ * @return Query transaction or {@code null}.
+ */
+ public static <T extends IgniteInternalTx> T queryTransaction(Context ctx, GridCacheSharedContext<?, ?> cctx) {
+ GridCacheVersion txId = queryTransactionVersion(ctx);
+
+ return txId == null ? null : cctx.tm().tx(txId);
+ }
+
+ /**
+ * @param ctx Context.
+ * @return Query transaction version if exists or {@code null}.
+ */
+ public static @Nullable GridCacheVersion queryTransactionVersion(Context ctx) {
+ return ctx.unwrap(GridCacheVersion.class);
+ }
}
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/QueryChecker.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/QueryChecker.java
index 2d8f936919d..a512d5cb365 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/QueryChecker.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/QueryChecker.java
@@ -27,21 +27,24 @@ import java.util.List;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
-
import org.apache.calcite.tools.FrameworkConfig;
import org.apache.ignite.Ignite;
import org.apache.ignite.cache.query.FieldsQueryCursor;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
-import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicCache;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheAdapter;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition;
+import org.apache.ignite.internal.processors.cache.transactions.TransactionProxyImpl;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.query.QueryContext;
import org.apache.ignite.internal.processors.query.QueryEngine;
+import org.apache.ignite.internal.processors.query.calcite.integration.AbstractBasicIntegrationTransactionalTest.SqlTransactionMode;
import org.apache.ignite.internal.processors.query.schema.management.SchemaManager;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.G;
import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.transactions.Transaction;
import org.hamcrest.CoreMatchers;
import org.hamcrest.CustomTypeSafeMatcher;
import org.hamcrest.Matcher;
@@ -274,6 +277,9 @@ public abstract class QueryChecker {
/** */
private final String qry;
+ /** */
+ private final Transaction tx;
+
/** */
private final ArrayList<Matcher<String>> planMatchers = new ArrayList<>();
@@ -303,7 +309,16 @@ public abstract class QueryChecker {
/** */
public QueryChecker(String qry) {
+ this(qry, null, SqlTransactionMode.NONE);
+ }
+
+ /** */
+ public QueryChecker(String qry, Transaction tx, SqlTransactionMode sqlTxMode) {
+ assert (tx != null && sqlTxMode != SqlTransactionMode.NONE)
+ || (tx == null && sqlTxMode == SqlTransactionMode.NONE) : "mode = " + sqlTxMode + ", tx = " + tx;
+
this.qry = qry;
+ this.tx = tx;
}
/** */
@@ -382,7 +397,11 @@ public abstract class QueryChecker {
// Check plan.
QueryEngine engine = getEngine();
- QueryContext ctx = frameworkCfg != null ? QueryContext.of(frameworkCfg) : null;
+ GridCacheVersion txVer = tx != null
+ ? ((TransactionProxyImpl)tx).tx().xidVersion()
+ : null;
+
+ QueryContext ctx = (frameworkCfg != null || txVer != null) ? QueryContext.of(frameworkCfg, txVer) : null;
List<FieldsQueryCursor<List<?>>> explainCursors =
engine.query(ctx, "PUBLIC", "EXPLAIN PLAN FOR " + qry, params);
@@ -391,7 +410,7 @@ public abstract class QueryChecker {
List<List<?>> explainRes = explainCursor.getAll();
String actualPlan = (String)explainRes.get(0).get(0);
- if (!F.isEmpty(planMatchers)) {
+ if (!F.isEmpty(planMatchers) && tx == null) {
for (Matcher<String> matcher : planMatchers)
assertThat("Invalid plan:\n" + actualPlan + "\n for query: " + qry, actualPlan, matcher);
}
@@ -520,7 +539,7 @@ public abstract class QueryChecker {
* @param cacheName Cache to check reservations.
*/
public static void awaitReservationsRelease(IgniteEx node, String cacheName) throws IgniteInterruptedCheckedException {
- GridDhtAtomicCache c = GridTestUtils.getFieldValue(node.cachex(cacheName), "delegate");
+ GridDhtCacheAdapter c = GridTestUtils.getFieldValue(node.cachex(cacheName), "delegate");
List<GridDhtLocalPartition> parts = c.topology().localPartitions();
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/KeyFilteringCursorTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/KeyFilteringCursorTest.java
new file mode 100644
index 00000000000..ff014e59f4e
--- /dev/null
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/KeyFilteringCursorTest.java
@@ -0,0 +1,160 @@
+/*
+ * 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.calcite.exec;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.cache.query.index.sorted.IndexKeyDefinition;
+import org.apache.ignite.internal.cache.query.index.sorted.IndexKeyTypeSettings;
+import org.apache.ignite.internal.cache.query.index.sorted.IndexRow;
+import org.apache.ignite.internal.cache.query.index.sorted.InlineIndexRowHandler;
+import org.apache.ignite.internal.cache.query.index.sorted.inline.InlineIndexKeyType;
+import org.apache.ignite.internal.cache.query.index.sorted.keys.IndexKey;
+import org.apache.ignite.internal.cache.query.index.sorted.keys.IntegerIndexKey;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.KeyCacheObjectImpl;
+import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
+import org.apache.ignite.internal.processors.cache.persistence.CacheDataRowAdapter;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.util.lang.GridCursor;
+import org.apache.ignite.internal.util.typedef.internal.CU;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+/** */
+public class KeyFilteringCursorTest {
+ /** */
+ @Test
+ public void testSimple() throws Exception {
+ assertEquals(Collections.emptyList(), all(cursor(), keys()));
+
+ assertEquals(Arrays.asList(1, 3), all(cursor(1, 2, 3), keys(2)));
+
+ assertEquals(Arrays.asList(1, 2, 3), all(cursor(1, 2, 3), keys()));
+
+ assertEquals(Collections.emptyList(), all(cursor(1, 2, 3), keys(1, 2, 3)));
+
+ assertEquals(Arrays.asList(1, 2, 3), all(cursor(1, 2, 3), keys(4, 5, 6)));
+
+ assertEquals(Arrays.asList(2, 3), all(cursor(1, 2, 3), keys(1, 4, 5, 6)));
+
+ assertEquals(Collections.emptyList(), all(cursor(1, 2, 3), keys(1, 2, 3, 4, 5, 6)));
+
+ assertEquals(Collections.singletonList(2), all(cursor(1, 2, 3, 4, 5, 6), keys(1, 3, 4, 5, 6)));
+
+ assertEquals(Arrays.asList(1, 2, 3, 4), all(cursor(1, 2, 3, 4, 5, 6), keys(5, 6)));
+ }
+
+ /** */
+ private List<Integer> all(GridCursor<IndexRow> rawCursor, Set<KeyCacheObject> skipKeys) throws IgniteCheckedException {
+ GridCursor<IndexRow> cursor = new KeyFilteringCursor<>(rawCursor, skipKeys, r -> r.cacheDataRow().key());
+
+ List<Integer> res = new ArrayList<>();
+
+ while (cursor.next())
+ res.add((Integer)cursor.get().key(0).key());
+
+ return res;
+ }
+
+ /** */
+ private GridCursor<IndexRow> cursor(int...nums) {
+ List<IndexRow> rows = new ArrayList<>();
+
+ for (int num : nums) {
+ rows.add(new IndexRow() {
+ @Override public IndexKey key(int idx) {
+ return new IntegerIndexKey(num);
+ }
+
+ @Override public int keysCount() {
+ return 1;
+ }
+
+ @Override public long link() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override public InlineIndexRowHandler rowHandler() {
+ return new InlineIndexRowHandler() {
+ @Override public IndexKey indexKey(int idx, CacheDataRow row) {
+ assert idx == 0;
+
+ return new IntegerIndexKey(num);
+ }
+
+ @Override public List<InlineIndexKeyType> inlineIndexKeyTypes() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override public List<IndexKeyDefinition> indexKeyDefinitions() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override public IndexKeyTypeSettings indexKeyTypeSettings() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override public int partition(CacheDataRow row) {
+ return row.partition();
+ }
+
+ @Override public Object cacheKey(CacheDataRow row) {
+ return row.key();
+ }
+
+ @Override public Object cacheValue(CacheDataRow row) {
+ return row.value();
+ }
+ };
+ }
+
+ @Override public CacheDataRow cacheDataRow() {
+ return new CacheDataRowAdapter(
+ KeyFilteringCursorTest.this.key(num),
+ null,
+ new GridCacheVersion(1, 1, 1L),
+ CU.EXPIRE_TIME_ETERNAL
+ );
+ }
+
+ @Override public boolean indexPlainRow() {
+ return false;
+ }
+ });
+ }
+
+ return new SortedListRangeCursor<>(null, rows, null, null, true, true);
+ }
+
+ /** */
+ private Set<KeyCacheObject> keys(int...nums) {
+ return Arrays.stream(nums).mapToObj(this::key).collect(Collectors.toSet());
+ }
+
+ /** */
+ private KeyCacheObjectImpl key(int num) {
+ return new KeyCacheObjectImpl(num, null, -1);
+ }
+}
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementorTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementorTest.java
index 209eb75195c..abc0268a3bc 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementorTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementorTest.java
@@ -126,6 +126,7 @@ public class LogicalRelImplementorTest extends GridCommonAbstractTest {
NoOpMemoryTracker.INSTANCE,
NoOpIoTracker.INSTANCE,
0,
+ null,
null
) {
@Override public ColocationGroup group(long srcId) {
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/RuntimeSortedIndexTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/RuntimeSortedIndexTest.java
index 818b1074fba..450e555cfb0 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/RuntimeSortedIndexTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/RuntimeSortedIndexTest.java
@@ -119,6 +119,7 @@ public class RuntimeSortedIndexTest extends GridCommonAbstractTest {
NoOpMemoryTracker.INSTANCE,
NoOpIoTracker.INSTANCE,
0,
+ null,
null),
RelCollations.of(ImmutableIntList.copyOf(idxCols)),
(o1, o2) -> {
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/AbstractExecutionTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/AbstractExecutionTest.java
index 8a259226bbe..0858936705b 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/AbstractExecutionTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/AbstractExecutionTest.java
@@ -307,7 +307,8 @@ public class AbstractExecutionTest extends GridCommonAbstractTest {
NoOpMemoryTracker.INSTANCE,
NoOpIoTracker.INSTANCE,
0,
- ImmutableMap.of()
+ ImmutableMap.of(),
+ null
);
}
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/AbstractBasicIntegrationTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/AbstractBasicIntegrationTest.java
index eed1ae56e85..d6d562a99ac 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/AbstractBasicIntegrationTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/AbstractBasicIntegrationTest.java
@@ -147,7 +147,12 @@ public class AbstractBasicIntegrationTest extends GridCommonAbstractTest {
/** */
protected List<List<?>> executeSql(String sql, Object... args) {
- CalciteQueryProcessor qryProc = Commons.lookupComponent(client.context(), CalciteQueryProcessor.class);
+ return executeSql(client, sql, args);
+ }
+
+ /** */
+ protected List<List<?>> executeSql(IgniteEx ignite, String sql, Object... args) {
+ CalciteQueryProcessor qryProc = Commons.lookupComponent(ignite.context(), CalciteQueryProcessor.class);
List<FieldsQueryCursor<List<?>>> cur = qryProc.query(queryContext(), "PUBLIC", sql, args);
@@ -174,12 +179,12 @@ public class AbstractBasicIntegrationTest extends GridCommonAbstractTest {
/** */
protected IgniteCache<Integer, Employer> createAndPopulateTable() {
- return createAndPopulateTable(2, CacheMode.PARTITIONED);
+ return createAndPopulateTable(client, 2, CacheMode.PARTITIONED);
}
/** */
- protected IgniteCache<Integer, Employer> createAndPopulateTable(int backups, CacheMode cacheMode) {
- IgniteCache<Integer, Employer> person = client.getOrCreateCache(new CacheConfiguration<Integer, Employer>()
+ protected IgniteCache<Integer, Employer> createAndPopulateTable(Ignite ignite, int backups, CacheMode cacheMode) {
+ IgniteCache<Integer, Employer> person = ignite.getOrCreateCache(this.<Integer, Employer>cacheConfiguration()
.setName(TABLE_NAME)
.setSqlSchema("PUBLIC")
.setQueryEntities(F.asList(new QueryEntity(Integer.class, Employer.class)
@@ -193,15 +198,25 @@ public class AbstractBasicIntegrationTest extends GridCommonAbstractTest {
int idx = 0;
- person.put(idx++, new Employer("Igor", 10d));
- person.put(idx++, new Employer(null, 15d));
- person.put(idx++, new Employer("Ilya", 15d));
- person.put(idx++, new Employer("Roma", 10d));
- person.put(idx++, new Employer("Roma", 10d));
+ put(ignite, person, idx++, new Employer("Igor", 10d));
+ put(ignite, person, idx++, new Employer(null, 15d));
+ put(ignite, person, idx++, new Employer("Ilya", 15d));
+ put(ignite, person, idx++, new Employer("Roma", 10d));
+ put(ignite, person, idx, new Employer("Roma", 10d));
return person;
}
+ /** */
+ protected <K, V> void put(Ignite ignite, IgniteCache<K, V> c, K key, V val) {
+ c.put(key, val);
+ }
+
+ /** */
+ protected <K, V> CacheConfiguration<K, V> cacheConfiguration() {
+ return new CacheConfiguration<>();
+ }
+
/** */
protected CalciteQueryProcessor queryProcessor(Ignite ignite) {
return Commons.lookupComponent(((IgniteEx)ignite).context(), CalciteQueryProcessor.class);
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/AbstractBasicIntegrationTransactionalTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/AbstractBasicIntegrationTransactionalTest.java
new file mode 100644
index 00000000000..58abf9d9629
--- /dev/null
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/AbstractBasicIntegrationTransactionalTest.java
@@ -0,0 +1,242 @@
+/*
+ * 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.calcite.integration;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.calcite.CalciteQueryEngineConfiguration;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.processors.cache.transactions.TransactionProxyImpl;
+import org.apache.ignite.internal.processors.query.QueryContext;
+import org.apache.ignite.internal.processors.query.QueryEngine;
+import org.apache.ignite.internal.processors.query.calcite.QueryChecker;
+import org.apache.ignite.internal.processors.query.calcite.util.Commons;
+import org.apache.ignite.transactions.Transaction;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
+import static org.apache.ignite.transactions.TransactionIsolation.READ_COMMITTED;
+
+/** */
+@RunWith(Parameterized.class)
+public abstract class AbstractBasicIntegrationTransactionalTest extends AbstractBasicIntegrationTest {
+ /** */
+ @Parameterized.Parameter()
+ public SqlTransactionMode sqlTxMode;
+
+ /** @return Test parameters. */
+ @Parameterized.Parameters(name = "sqlTxMode={0}")
+ public static Collection<?> parameters() {
+ return Arrays.asList(SqlTransactionMode.values());
+ }
+
+ /** */
+ protected static SqlTransactionMode currentMode;
+
+ /** */
+ protected static Transaction tx;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.getSqlConfiguration().setQueryEnginesConfiguration(new CalciteQueryEngineConfiguration());
+ cfg.getTransactionConfiguration().setTxAwareQueriesEnabled(true);
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ if (currentMode != null && sqlTxMode == currentMode)
+ return;
+
+ currentMode = sqlTxMode;
+
+ clearTransaction();
+
+ stopAllGrids();
+
+ init();
+ }
+
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ clearTransaction();
+
+ super.afterTest();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ clearTransaction();
+
+ stopAllGrids();
+
+ currentMode = null;
+
+ tx = null;
+
+ super.afterTestsStopped();
+ }
+
+ /** */
+ protected void init() throws Exception {
+ startGrids(nodeCount());
+
+ client = startClientGrid("client");
+ }
+
+ /** {@inheritDoc} */
+ @Override protected QueryChecker assertQuery(IgniteEx ignite, String qry) {
+ return new QueryChecker(qry, tx, sqlTxMode) {
+ @Override protected QueryEngine getEngine() {
+ return Commons.lookupComponent(ignite.context(), QueryEngine.class);
+ }
+ };
+ }
+
+ /** {@inheritDoc} */
+ @Override protected List<List<?>> executeSql(IgniteEx ignite, String sql, Object... args) {
+ if (sqlTxMode != SqlTransactionMode.NONE && tx == null)
+ startTransaction(ignite);
+
+ return super.executeSql(ignite, sql, args);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected QueryContext queryContext() {
+ return QueryContext.of(tx != null ? ((TransactionProxyImpl<?, ?>)tx).tx().xidVersion() : null);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected List<List<?>> sql(IgniteEx ignite, String sql, Object... params) {
+ if (sqlTxMode != SqlTransactionMode.NONE && tx == null)
+ startTransaction(ignite);
+
+ return super.sql(ignite, sql, params);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected <K, V> void put(Ignite node, IgniteCache<K, V> cache, K key, V val) {
+ invokeAction(node, () -> {
+ cache.put(key, val);
+ return null;
+ });
+ }
+
+ /** {@inheritDoc} */
+ @Override protected <K, V> CacheConfiguration<K, V> cacheConfiguration() {
+ return super.<K, V>cacheConfiguration().setAtomicityMode(sqlTxMode == SqlTransactionMode.NONE
+ ? CacheAtomicityMode.ATOMIC
+ : CacheAtomicityMode.TRANSACTIONAL);
+ }
+
+ /** */
+ protected <T> T invokeAction(Ignite node, SupplierX<T> action) {
+ if (tx == null && sqlTxMode != SqlTransactionMode.NONE)
+ startTransaction(node);
+
+ switch (sqlTxMode) {
+ case ALL:
+ return txAction(node, action);
+ case NONE:
+ return action.get();
+ case RANDOM:
+ if (ThreadLocalRandom.current().nextBoolean())
+ return action.get();
+ else
+ return txAction(node, action);
+ default:
+ throw new IllegalArgumentException();
+ }
+ }
+
+ /** */
+ public <T> T txAction(Ignite node, SupplierX<T> action) {
+ tx.resume();
+
+ try {
+ return action.get();
+ }
+ finally {
+ tx.suspend();
+ }
+ }
+
+ /** */
+ protected void startTransaction(Ignite node) {
+ tx = node.transactions().txStart(PESSIMISTIC, READ_COMMITTED, getTestTimeout(), 100);
+
+ tx.suspend();
+ }
+
+ /** */
+ protected void clearTransaction() {
+ if (tx == null)
+ return;
+
+ tx.resume();
+
+ tx.rollback();
+
+ tx = null;
+ }
+
+ /** */
+ public String atomicity() {
+ return "atomicity=" + (sqlTxMode == SqlTransactionMode.NONE ? CacheAtomicityMode.ATOMIC : CacheAtomicityMode.TRANSACTIONAL);
+ }
+
+ /** */
+ public interface SupplierX<T> {
+ /** */
+ T getx() throws Exception;
+
+ /** */
+ default T get() {
+ try {
+ return getx();
+ }
+ catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ /** */
+ public enum SqlTransactionMode {
+ /** All put, remove and SQL dml will be executed inside transaction. */
+ ALL,
+
+ /** Only some DML operations will be executed inside transaction. */
+ RANDOM,
+
+ /** Don't use transaction for DML. */
+ NONE
+ }
+}
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/AggregatesIntegrationTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/AggregatesIntegrationTest.java
index 4b910029cae..75711fa95a8 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/AggregatesIntegrationTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/AggregatesIntegrationTest.java
@@ -17,13 +17,14 @@
package org.apache.ignite.internal.processors.query.calcite.integration;
+import java.util.Collections;
import java.util.List;
import org.apache.ignite.IgniteCache;
-import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.QueryEntity;
+import org.apache.ignite.cache.QueryIndex;
import org.apache.ignite.cache.query.annotations.QuerySqlField;
-import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.internal.processors.query.QueryUtils;
import org.apache.ignite.internal.processors.query.calcite.QueryChecker;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.testframework.GridTestUtils;
@@ -32,7 +33,7 @@ import org.junit.Test;
/**
*
*/
-public class AggregatesIntegrationTest extends AbstractBasicIntegrationTest {
+public class AggregatesIntegrationTest extends AbstractBasicIntegrationTransactionalTest {
/** */
@Test
public void testMinMaxWithTable() {
@@ -48,7 +49,7 @@ public class AggregatesIntegrationTest extends AbstractBasicIntegrationTest {
for (String idx : indexes) {
for (int backups = -1; backups < 3; ++backups) {
executeSql("create table tbl(id integer primary key, val0 integer, val1 float, val2 varchar) " +
- "with template=" + (backups < 0 ? "replicated" : "partitioned,backups=" + backups));
+ "with template=" + (backups < 0 ? "replicated" : "partitioned,backups=" + backups) + "," + atomicity());
executeSql("create index test_idx on tbl(" + idx + ")");
@@ -59,6 +60,8 @@ public class AggregatesIntegrationTest extends AbstractBasicIntegrationTest {
assertQuery("select max(val0) from tbl").returns(5).check();
assertQuery("select max(val1) from tbl").returns(50.0f).check();
+ clearTransaction();
+
executeSql("drop table tbl");
}
}
@@ -88,6 +91,8 @@ public class AggregatesIntegrationTest extends AbstractBasicIntegrationTest {
assertQuery("select max(salary) from person").returns(15.0).check();
assertQuery("select max(descVal) from person").returns(15.0).check();
+ clearTransaction();
+
client.destroyCache(TABLE_NAME);
}
}
@@ -100,6 +105,8 @@ public class AggregatesIntegrationTest extends AbstractBasicIntegrationTest {
assertQuery("select count(*) from person").returns(7L).check();
+ clearTransaction();
+
client.destroyCache(TABLE_NAME);
}
@@ -111,6 +118,11 @@ public class AggregatesIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testCountIndexedField() {
+ // Check count with two columns index.
+ sql("CREATE TABLE tbl (a INT, b INT, c INT) WITH " + atomicity());
+ sql("CREATE INDEX idx_a ON tbl(a, c)");
+ sql("CREATE INDEX idx_b ON tbl(b DESC, c)");
+
createAndPopulateIndexedTable(1, CacheMode.PARTITIONED);
assertQuery("select count(salary) from person").returns(4L).check();
@@ -126,11 +138,6 @@ public class AggregatesIntegrationTest extends AbstractBasicIntegrationTest {
.returns(1L, null)
.check();
- // Check count with two columns index.
- sql("CREATE TABLE tbl (a INT, b INT, c INT)");
- sql("CREATE INDEX idx_a ON tbl(a, c)");
- sql("CREATE INDEX idx_b ON tbl(b DESC, c)");
-
for (int i = 0; i < 100; i++) {
sql("INSERT INTO tbl VALUES (null, null, ?)", i % 2 == 0 ? i : null);
sql("INSERT INTO tbl VALUES (?, ?, ?)", i, i, i % 2 == 0 ? null : i);
@@ -221,7 +228,7 @@ public class AggregatesIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@SuppressWarnings("ThrowableNotThrown")
@Test
- public void testMultipleRowsFromSingleAggr() throws IgniteCheckedException {
+ public void testMultipleRowsFromSingleAggr() {
createAndPopulateTable();
GridTestUtils.assertThrowsWithCause(() -> assertQuery("SELECT (SELECT name FROM person)").check(),
@@ -240,10 +247,12 @@ public class AggregatesIntegrationTest extends AbstractBasicIntegrationTest {
IgniteCache<Integer, Employer> person = client.cache(cacheName);
+ clearTransaction();
+
person.clear();
for (int gridIdx = 0; gridIdx < nodeCount(); gridIdx++)
- person.put(primaryKey(grid(gridIdx).cache(cacheName)), new Employer(gridIdx == 0 ? "Emp" : null, 0.0d));
+ put(client, person, primaryKey(grid(gridIdx).cache(cacheName)), new Employer(gridIdx == 0 ? "Emp" : null, 0.0d));
GridTestUtils.assertThrowsWithCause(() -> assertQuery("SELECT (SELECT name FROM person)").check(),
IllegalArgumentException.class);
@@ -280,12 +289,12 @@ public class AggregatesIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
- public void testColocatedAggregate() throws Exception {
+ public void testColocatedAggregate() {
executeSql("CREATE TABLE t1(id INT, val0 VARCHAR, val1 VARCHAR, val2 VARCHAR, PRIMARY KEY(id, val1)) " +
- "WITH AFFINITY_KEY=val1");
+ "WITH AFFINITY_KEY=val1," + atomicity());
executeSql("CREATE TABLE t2(id INT, val0 VARCHAR, val1 VARCHAR, val2 VARCHAR, PRIMARY KEY(id, val1)) " +
- "WITH AFFINITY_KEY=val1");
+ "WITH AFFINITY_KEY=val1," + atomicity());
for (int i = 0; i < 100; i++)
executeSql("INSERT INTO t1 VALUES (?, ?, ?, ?)", i, "val" + i, "val" + i % 2, "val" + i);
@@ -312,8 +321,8 @@ public class AggregatesIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
- public void testEverySomeAggregate() throws Exception {
- executeSql("CREATE TABLE t(c1 INT, c2 INT)");
+ public void testEverySomeAggregate() {
+ executeSql("CREATE TABLE t(c1 INT, c2 INT) WITH " + atomicity());
executeSql("INSERT INTO t VALUES (null, 0)");
executeSql("INSERT INTO t VALUES (0, null)");
executeSql("INSERT INTO t VALUES (null, null)");
@@ -328,9 +337,32 @@ public class AggregatesIntegrationTest extends AbstractBasicIntegrationTest {
assertQuery("SELECT SOME(c1 > c2) FROM t").returns(false).check();
}
+ /** */
+ @Test
+ public void testCountIndexedFieldSegmented() {
+ client.getOrCreateCache(cacheConfiguration()
+ .setName("cache_seg")
+ .setSqlSchema(QueryUtils.DFLT_SCHEMA)
+ .setQueryEntities(Collections.singleton(
+ new QueryEntity()
+ .setKeyType(Integer.class.getName())
+ .setValueType("tbl_seg")
+ .setTableName("TBL_SEG")
+ .addQueryField("a", Integer.class.getName(), null)
+ .addQueryField("b", Integer.class.getName(), null)
+ .setIndexes(F.asList(new QueryIndex("a", true), new QueryIndex("b", false)))))
+ .setQueryParallelism(5));
+
+ for (int i = 0; i < 100; i++)
+ sql("INSERT INTO tbl_seg (_key, a, b) VALUES (?, ?, ?)", i, i % 2 == 0 ? i : null, i % 2 == 0 ? null : i);
+
+ assertQuery("SELECT COUNT(a) FROM tbl_seg").returns(50L).check();
+ assertQuery("SELECT COUNT(b) FROM tbl_seg").returns(50L).check();
+ }
+
/** */
protected void createAndPopulateIndexedTable(int backups, CacheMode cacheMode) {
- IgniteCache<Integer, IndexedEmployer> person = client.getOrCreateCache(new CacheConfiguration<Integer, IndexedEmployer>()
+ IgniteCache<Integer, IndexedEmployer> person = client.getOrCreateCache(this.<Integer, IndexedEmployer>cacheConfiguration()
.setName(TABLE_NAME)
.setSqlSchema("PUBLIC")
.setQueryEntities(F.asList(new QueryEntity(Integer.class, IndexedEmployer.class).setTableName(TABLE_NAME)))
@@ -340,13 +372,13 @@ public class AggregatesIntegrationTest extends AbstractBasicIntegrationTest {
int idx = 0;
- person.put(idx++, new IndexedEmployer("Igor", 5d, 9d));
- person.put(idx++, new IndexedEmployer(null, 3d, null));
- person.put(idx++, new IndexedEmployer("Ilya", 1d, 1d));
- person.put(idx++, new IndexedEmployer("Roma", null, 9d));
- person.put(idx++, new IndexedEmployer(null, null, null));
- person.put(idx++, new IndexedEmployer("Oleg", 15d, 15d));
- person.put(idx++, new IndexedEmployer("Maya", null, null));
+ put(client, person, idx++, new IndexedEmployer("Igor", 5d, 9d));
+ put(client, person, idx++, new IndexedEmployer(null, 3d, null));
+ put(client, person, idx++, new IndexedEmployer("Ilya", 1d, 1d));
+ put(client, person, idx++, new IndexedEmployer("Roma", null, 9d));
+ put(client, person, idx++, new IndexedEmployer(null, null, null));
+ put(client, person, idx++, new IndexedEmployer("Oleg", 15d, 15d));
+ put(client, person, idx, new IndexedEmployer("Maya", null, null));
}
/** */
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/CalciteBasicSecondaryIndexIntegrationTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/CalciteBasicSecondaryIndexIntegrationTest.java
index 43f6fbb496c..28146315298 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/CalciteBasicSecondaryIndexIntegrationTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/CalciteBasicSecondaryIndexIntegrationTest.java
@@ -46,7 +46,7 @@ import static org.hamcrest.CoreMatchers.not;
/**
* Basic index tests.
*/
-public class CalciteBasicSecondaryIndexIntegrationTest extends AbstractBasicIntegrationTest {
+public class CalciteBasicSecondaryIndexIntegrationTest extends AbstractBasicIntegrationTransactionalTest {
/** */
private static final String PK_IDX_NAME = QueryUtils.PRIMARY_KEY_INDEX;
@@ -69,8 +69,8 @@ public class CalciteBasicSecondaryIndexIntegrationTest extends AbstractBasicInte
private static final String NAME_DATE_IDX = "NAME_DATE_IDX";
/** {@inheritDoc} */
- @Override protected void beforeTestsStarted() throws Exception {
- super.beforeTestsStarted();
+ @Override protected void init() throws Exception {
+ super.init();
QueryEntity projEntity = new QueryEntity();
projEntity.setKeyType(Integer.class.getName());
@@ -105,35 +105,6 @@ public class CalciteBasicSecondaryIndexIntegrationTest extends AbstractBasicInte
IgniteCache<Integer, Developer> devCache = client.createCache(projCfg);
- devCache.put(1, new Developer("Mozart", 3, "Vienna", 33));
- devCache.put(2, new Developer("Beethoven", 2, "Vienna", 44));
- devCache.put(3, new Developer("Bach", 1, "Leipzig", 55));
- devCache.put(4, new Developer("Strauss", 2, "Munich", 66));
-
- devCache.put(5, new Developer("Vagner", 4, "Leipzig", 70));
- devCache.put(6, new Developer("Chaikovsky", 5, "Votkinsk", 53));
- devCache.put(7, new Developer("Verdy", 6, "Rankola", 88));
- devCache.put(8, new Developer("Stravinsky", 7, "Spt", 89));
- devCache.put(9, new Developer("Rahmaninov", 8, "Starorussky ud", 70));
- devCache.put(10, new Developer("Shubert", 9, "Vienna", 31));
- devCache.put(11, new Developer("Glinka", 10, "Smolenskaya gb", 53));
-
- devCache.put(12, new Developer("Einaudi", 11, "", -1));
- devCache.put(13, new Developer("Glass", 12, "", -1));
- devCache.put(14, new Developer("Rihter", 13, "", -1));
-
- devCache.put(15, new Developer("Marradi", 14, "", -1));
- devCache.put(16, new Developer("Zimmer", 15, "", -1));
- devCache.put(17, new Developer("Hasaishi", 16, "", -1));
-
- devCache.put(18, new Developer("Arnalds", 17, "", -1));
- devCache.put(19, new Developer("Yiruma", 18, "", -1));
- devCache.put(20, new Developer("O'Halloran", 19, "", -1));
-
- devCache.put(21, new Developer("Cacciapaglia", 20, "", -1));
- devCache.put(22, new Developer("Prokofiev", 21, "", -1));
- devCache.put(23, new Developer("Musorgskii", 22, "", -1));
-
QueryEntity bdEntity = new QueryEntity();
bdEntity.setKeyType(Integer.class.getName());
bdEntity.setKeyFieldName("id");
@@ -158,16 +129,8 @@ public class CalciteBasicSecondaryIndexIntegrationTest extends AbstractBasicInte
IgniteCache<Integer, Birthday> bdCache = client.createCache(bdCfg);
- bdCache.put(1, new Birthday("Mozart", Date.valueOf("1756-01-27")));
- bdCache.put(2, new Birthday("Beethoven", null));
- bdCache.put(3, new Birthday("Bach", Date.valueOf("1685-03-31")));
- bdCache.put(4, new Birthday("Strauss", Date.valueOf("1864-06-11")));
- bdCache.put(5, new Birthday("Vagner", Date.valueOf("1813-05-22")));
- bdCache.put(6, new Birthday("Chaikovsky", Date.valueOf("1840-05-07")));
- bdCache.put(7, new Birthday("Verdy", Date.valueOf("1813-10-10")));
-
IgniteCache<CalciteQueryProcessorTest.Key, CalciteQueryProcessorTest.Developer> tblWithAff =
- client.getOrCreateCache(new CacheConfiguration<CalciteQueryProcessorTest.Key, CalciteQueryProcessorTest.Developer>()
+ client.getOrCreateCache(this.<CalciteQueryProcessorTest.Key, CalciteQueryProcessorTest.Developer>cacheConfiguration()
.setName("TBL_WITH_AFF_KEY")
.setSqlSchema("PUBLIC")
.setBackups(1)
@@ -175,11 +138,8 @@ public class CalciteBasicSecondaryIndexIntegrationTest extends AbstractBasicInte
.setTableName("TBL_WITH_AFF_KEY")))
);
- tblWithAff.put(new CalciteQueryProcessorTest.Key(1, 2), new CalciteQueryProcessorTest.Developer("Petr", 10));
- tblWithAff.put(new CalciteQueryProcessorTest.Key(2, 3), new CalciteQueryProcessorTest.Developer("Ivan", 11));
-
IgniteCache<Integer, CalciteQueryProcessorTest.Developer> tblConstrPk =
- client.getOrCreateCache(new CacheConfiguration<Integer, CalciteQueryProcessorTest.Developer>()
+ client.getOrCreateCache(this.<Integer, CalciteQueryProcessorTest.Developer>cacheConfiguration()
.setName("TBL_CONSTR_PK")
.setSqlSchema("PUBLIC")
.setBackups(0)
@@ -189,11 +149,51 @@ public class CalciteBasicSecondaryIndexIntegrationTest extends AbstractBasicInte
.addQueryField("id", Integer.class.getName(), null)))
);
- tblConstrPk.put(1, new CalciteQueryProcessorTest.Developer("Petr", 10));
- tblConstrPk.put(2, new CalciteQueryProcessorTest.Developer("Ivan", 11));
-
executeSql("CREATE TABLE PUBLIC.UNWRAP_PK" + " (F1 VARCHAR, F2 BIGINT, F3 BIGINT, F4 BIGINT, " +
- "CONSTRAINT PK PRIMARY KEY (F2, F1)) WITH \"backups=0, affinity_key=F1\"");
+ "CONSTRAINT PK PRIMARY KEY (F2, F1)) WITH \"backups=0, affinity_key=F1," + atomicity() + "\"");
+
+ put(client, devCache, 1, new Developer("Mozart", 3, "Vienna", 33));
+ put(client, devCache, 2, new Developer("Beethoven", 2, "Vienna", 44));
+ put(client, devCache, 3, new Developer("Bach", 1, "Leipzig", 55));
+ put(client, devCache, 4, new Developer("Strauss", 2, "Munich", 66));
+
+ put(client, devCache, 5, new Developer("Vagner", 4, "Leipzig", 70));
+ put(client, devCache, 6, new Developer("Chaikovsky", 5, "Votkinsk", 53));
+ put(client, devCache, 7, new Developer("Verdy", 6, "Rankola", 88));
+ put(client, devCache, 8, new Developer("Stravinsky", 7, "Spt", 89));
+ put(client, devCache, 9, new Developer("Rahmaninov", 8, "Starorussky ud", 70));
+ put(client, devCache, 10, new Developer("Shubert", 9, "Vienna", 31));
+ put(client, devCache, 11, new Developer("Glinka", 10, "Smolenskaya gb", 53));
+
+ put(client, devCache, 12, new Developer("Einaudi", 11, "", -1));
+ put(client, devCache, 13, new Developer("Glass", 12, "", -1));
+ put(client, devCache, 14, new Developer("Rihter", 13, "", -1));
+
+ put(client, devCache, 15, new Developer("Marradi", 14, "", -1));
+ put(client, devCache, 16, new Developer("Zimmer", 15, "", -1));
+ put(client, devCache, 17, new Developer("Hasaishi", 16, "", -1));
+
+ put(client, devCache, 18, new Developer("Arnalds", 17, "", -1));
+ put(client, devCache, 19, new Developer("Yiruma", 18, "", -1));
+ put(client, devCache, 20, new Developer("O'Halloran", 19, "", -1));
+
+ put(client, devCache, 21, new Developer("Cacciapaglia", 20, "", -1));
+ put(client, devCache, 22, new Developer("Prokofiev", 21, "", -1));
+ put(client, devCache, 23, new Developer("Musorgskii", 22, "", -1));
+
+ put(client, bdCache, 1, new Birthday("Mozart", Date.valueOf("1756-01-27")));
+ put(client, bdCache, 2, new Birthday("Beethoven", null));
+ put(client, bdCache, 3, new Birthday("Bach", Date.valueOf("1685-03-31")));
+ put(client, bdCache, 4, new Birthday("Strauss", Date.valueOf("1864-06-11")));
+ put(client, bdCache, 5, new Birthday("Vagner", Date.valueOf("1813-05-22")));
+ put(client, bdCache, 6, new Birthday("Chaikovsky", Date.valueOf("1840-05-07")));
+ put(client, bdCache, 7, new Birthday("Verdy", Date.valueOf("1813-10-10")));
+
+ put(client, tblWithAff, new CalciteQueryProcessorTest.Key(1, 2), new CalciteQueryProcessorTest.Developer("Petr", 10));
+ put(client, tblWithAff, new CalciteQueryProcessorTest.Key(2, 3), new CalciteQueryProcessorTest.Developer("Ivan", 11));
+
+ put(client, tblConstrPk, 1, new CalciteQueryProcessorTest.Developer("Petr", 10));
+ put(client, tblConstrPk, 2, new CalciteQueryProcessorTest.Developer("Ivan", 11));
executeSql("INSERT INTO PUBLIC.UNWRAP_PK(F1, F2, F3, F4) values ('Petr', 1, 2, 3)");
executeSql("INSERT INTO PUBLIC.UNWRAP_PK(F1, F2, F3, F4) values ('Ivan', 2, 2, 4)");
@@ -219,7 +219,7 @@ public class CalciteBasicSecondaryIndexIntegrationTest extends AbstractBasicInte
/** */
private <K, V> CacheConfiguration<K, V> cache(QueryEntity ent) {
- return new CacheConfiguration<K, V>(ent.getTableName())
+ return this.<K, V>cacheConfiguration().setName(ent.getTableName())
.setCacheMode(CacheMode.PARTITIONED)
.setBackups(1)
.setQueryEntities(singletonList(ent))
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/CorrelatesIntegrationTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/CorrelatesIntegrationTest.java
index 3e144df2800..652c2ecf0bf 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/CorrelatesIntegrationTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/CorrelatesIntegrationTest.java
@@ -23,13 +23,13 @@ import org.junit.Test;
/**
* Tests correlated queries.
*/
-public class CorrelatesIntegrationTest extends AbstractBasicIntegrationTest {
+public class CorrelatesIntegrationTest extends AbstractBasicIntegrationTransactionalTest {
/**
* Checks correlates are assigned before access.
*/
@Test
public void testCorrelatesAssignedBeforeAccess() {
- sql("create table test_tbl(v INTEGER)");
+ sql("create table test_tbl(v INTEGER) WITH " + atomicity());
sql("INSERT INTO test_tbl VALUES (1)");
assertQuery("SELECT t0.v, (SELECT t0.v + t1.v FROM test_tbl t1) AS j FROM test_tbl t0")
@@ -42,7 +42,7 @@ public class CorrelatesIntegrationTest extends AbstractBasicIntegrationTest {
*/
@Test
public void testCorrelatesWithTableSpool() {
- sql("CREATE TABLE test(i1 INT, i2 INT)");
+ sql("CREATE TABLE test(i1 INT, i2 INT) WITH " + atomicity());
sql("INSERT INTO test VALUES (1, 1), (2, 2)");
assertQuery("SELECT (SELECT t1.i1 + t1.i2 + t0.i2 FROM test t1 WHERE i1 = 1) FROM test t0")
@@ -57,9 +57,9 @@ public class CorrelatesIntegrationTest extends AbstractBasicIntegrationTest {
*/
@Test
public void testCorrelatesCollision() {
- sql("CREATE TABLE test1 (a INTEGER, b INTEGER)");
+ sql("CREATE TABLE test1 (a INTEGER, b INTEGER) WITH " + atomicity());
+ sql("CREATE TABLE test2 (a INTEGER, c INTEGER) WITH " + atomicity());
sql("INSERT INTO test1 VALUES (11, 1), (12, 2), (13, 3)");
- sql("CREATE TABLE test2 (a INTEGER, c INTEGER)");
sql("INSERT INTO test2 VALUES (11, 1), (12, 1), (13, 4)");
// Collision by correlate variables in the left hand.
@@ -82,9 +82,9 @@ public class CorrelatesIntegrationTest extends AbstractBasicIntegrationTest {
*/
@Test
public void testCorrelatedDistribution() {
- sql("CREATE TABLE dept(deptid INTEGER, name VARCHAR, PRIMARY KEY(deptid))");
+ sql("CREATE TABLE dept(deptid INTEGER, name VARCHAR, PRIMARY KEY(deptid)) WITH " + atomicity());
sql("CREATE TABLE emp(empid INTEGER, deptid INTEGER, name VARCHAR, PRIMARY KEY(empid, deptid)) " +
- "WITH AFFINITY_KEY=deptid");
+ "WITH AFFINITY_KEY=deptid," + atomicity());
sql("INSERT INTO dept VALUES (0, 'dept0'), (1, 'dept1'), (2, 'dept2')");
sql("INSERT INTO emp VALUES (0, 0, 'emp0'), (1, 0, 'emp1'), (2, 0, 'emp2'), " +
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/DataTypesTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/DataTypesTest.java
index 23474698fe5..4dc3d9cde3e 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/DataTypesTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/DataTypesTest.java
@@ -28,79 +28,74 @@ import org.apache.calcite.tools.FrameworkConfig;
import org.apache.calcite.tools.Frameworks;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.cache.QueryEntity;
-import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.internal.processors.query.IgniteSQLException;
import org.apache.ignite.internal.util.typedef.F;
import org.junit.Test;
import static org.apache.ignite.internal.processors.query.calcite.CalciteQueryProcessor.FRAMEWORK_CONFIG;
+import static org.junit.Assume.assumeTrue;
/**
* Test SQL data types.
*/
-public class DataTypesTest extends AbstractBasicIntegrationTest {
+public class DataTypesTest extends AbstractBasicIntegrationTransactionalTest {
/** Tests Other type. */
@Test
public void testOtherType() {
- try {
- executeSql("CREATE TABLE t(id INT, oth OTHER)");
-
- assertThrows("CREATE TABLE t2(id INT, oth OTHER DEFAULT 'str')", IgniteSQLException.class,
- "Type 'OTHER' doesn't support default value.");
-
- executeSql("INSERT INTO t VALUES (1, 'str')");
- executeSql("INSERT INTO t VALUES (2, 22)");
- executeSql("INSERT INTO t VALUES (3, CAST('33.5' AS FLOAT))");
- executeSql("INSERT INTO t VALUES (4, CAST('44.5' AS DOUBLE))");
- executeSql("INSERT INTO t VALUES (5, CAST('fd10556e-fc27-4a99-b5e4-89b8344cb3ce' as UUID))");
- executeSql("INSERT INTO t VALUES (6, NULL)");
- executeSql("INSERT INTO t VALUES (7, NULL)");
-
- assertQuery("SELECT oth FROM t order by id")
- .ordered()
- .returns("str")
- .returns(22)
- .returns(33.5f)
- .returns(44.5d)
- .returns(UUID.fromString("fd10556e-fc27-4a99-b5e4-89b8344cb3ce"))
- .returns(new Object[] {null})
- .returns(new Object[] {null})
- .check();
-
- assertThrows("SELECT MIN(oth) FROM t", UnsupportedOperationException.class,
- "MIN() is not supported for type 'OTHER'.");
-
- assertThrows("SELECT MAX(oth) FROM t", UnsupportedOperationException.class,
- "MAX() is not supported for type 'OTHER'.");
-
- assertThrows("SELECT AVG(oth) from t", UnsupportedOperationException.class,
- "AVG() is not supported for type 'OTHER'.");
-
- assertThrows("SELECT oth from t WHERE oth > 0", CalciteException.class,
- "Invalid types for comparison");
-
- assertQuery("SELECT oth FROM t WHERE oth=22")
- .returns(22)
- .check();
-
- assertQuery("SELECT oth FROM t WHERE oth is NULL")
- .returns(new Object[] {null})
- .returns(new Object[] {null})
- .check();
-
- executeSql("DELETE FROM t WHERE id IN (2, 3, 4, 5)");
-
- assertQuery("SELECT oth FROM t WHERE oth is not NULL")
- .returns("str")
- .check();
-
- assertQuery("SELECT oth FROM t WHERE oth!=22")
- .returns("str")
- .check();
- }
- finally {
- executeSql("DROP TABLE IF EXISTS t");
- }
+ executeSql("CREATE TABLE t(id INT, oth OTHER) WITH " + atomicity());
+
+ assertThrows("CREATE TABLE t2(id INT, oth OTHER DEFAULT 'str')", IgniteSQLException.class,
+ "Type 'OTHER' doesn't support default value.");
+
+ executeSql("INSERT INTO t VALUES (1, 'str')");
+ executeSql("INSERT INTO t VALUES (2, 22)");
+ executeSql("INSERT INTO t VALUES (3, CAST('33.5' AS FLOAT))");
+ executeSql("INSERT INTO t VALUES (4, CAST('44.5' AS DOUBLE))");
+ executeSql("INSERT INTO t VALUES (5, CAST('fd10556e-fc27-4a99-b5e4-89b8344cb3ce' as UUID))");
+ executeSql("INSERT INTO t VALUES (6, NULL)");
+ executeSql("INSERT INTO t VALUES (7, NULL)");
+
+ assertQuery("SELECT oth FROM t order by id")
+ .ordered()
+ .returns("str")
+ .returns(22)
+ .returns(33.5f)
+ .returns(44.5d)
+ .returns(UUID.fromString("fd10556e-fc27-4a99-b5e4-89b8344cb3ce"))
+ .returns(new Object[]{null})
+ .returns(new Object[]{null})
+ .check();
+
+ assertThrows("SELECT MIN(oth) FROM t", UnsupportedOperationException.class,
+ "MIN() is not supported for type 'OTHER'.");
+
+ assertThrows("SELECT MAX(oth) FROM t", UnsupportedOperationException.class,
+ "MAX() is not supported for type 'OTHER'.");
+
+ assertThrows("SELECT AVG(oth) from t", UnsupportedOperationException.class,
+ "AVG() is not supported for type 'OTHER'.");
+
+ assertThrows("SELECT oth from t WHERE oth > 0", CalciteException.class,
+ "Invalid types for comparison");
+
+ assertQuery("SELECT oth FROM t WHERE oth=22")
+ .returns(22)
+ .check();
+
+ assertQuery("SELECT oth FROM t WHERE oth is NULL")
+ .returns(new Object[]{null})
+ .returns(new Object[]{null})
+ .check();
+
+ executeSql("DELETE FROM t WHERE id IN (2, 3, 4, 5)");
+
+ assertQuery("SELECT oth FROM t WHERE oth is not NULL")
+ .returns("str")
+ .check();
+
+ assertQuery("SELECT oth FROM t WHERE oth!=22")
+ .returns("str")
+ .check();
}
/** Tests UUID without index. */
@@ -117,92 +112,87 @@ public class DataTypesTest extends AbstractBasicIntegrationTest {
/** Tests UUID type. */
private void testUuid(boolean indexed) {
- try {
- executeSql("CREATE TABLE t(id INT, name VARCHAR(255), uid UUID, primary key (id))");
-
- if (indexed)
- executeSql("CREATE INDEX uuid_idx ON t (uid);");
-
- UUID uuid1 = UUID.randomUUID();
- UUID uuid2 = UUID.randomUUID();
-
- UUID max = uuid1.compareTo(uuid2) > 0 ? uuid1 : uuid2;
- UUID min = max == uuid1 ? uuid2 : uuid1;
-
- executeSql("INSERT INTO t VALUES (1, 'fd10556e-fc27-4a99-b5e4-89b8344cb3ce', '" + uuid1 + "')");
- // Name == UUID
- executeSql("INSERT INTO t VALUES (2, '" + uuid2 + "', '" + uuid2 + "')");
- executeSql("INSERT INTO t VALUES (3, NULL, NULL)");
-
- assertQuery("SELECT * FROM t")
- .returns(1, "fd10556e-fc27-4a99-b5e4-89b8344cb3ce", uuid1)
- .returns(2, uuid2.toString(), uuid2)
- .returns(3, null, null)
- .check();
-
- assertQuery("SELECT uid FROM t WHERE uid < '" + max + "'")
- .returns(min)
- .check();
-
- assertQuery("SELECT uid FROM t WHERE '" + max + "' > uid")
- .returns(min)
- .check();
-
- assertQuery("SELECT * FROM t WHERE name = uid")
- .returns(2, uuid2.toString(), uuid2)
- .check();
-
- assertQuery("SELECT * FROM t WHERE name != uid")
- .returns(1, "fd10556e-fc27-4a99-b5e4-89b8344cb3ce", uuid1)
- .check();
-
- assertQuery("SELECT count(*), uid FROM t group by uid order by uid")
- .returns(1L, min)
- .returns(1L, max)
- .returns(1L, null)
- .check();
-
- assertQuery("SELECT count(*), uid FROM t group by uid having uid = " +
- "'" + uuid1 + "' order by uid")
- .returns(1L, uuid1)
- .check();
-
- assertQuery("SELECT t1.* from t t1, (select * from t) t2 where t1.uid = t2.name")
- .returns(2, uuid2.toString(), uuid2)
- .check();
-
- assertQuery("SELECT t1.* from t t1, (select * from t) t2 " +
- "where t1.uid = t2.uid")
- .returns(1, "fd10556e-fc27-4a99-b5e4-89b8344cb3ce", uuid1)
- .returns(2, uuid2.toString(), uuid2)
- .check();
-
- assertQuery("SELECT max(uid) from t")
- .returns(max)
- .check();
-
- assertQuery("SELECT min(uid) from t")
- .returns(min)
- .check();
-
- assertQuery("SELECT uid from t where uid is not NULL order by uid")
- .returns(min)
- .returns(max)
- .check();
-
- assertQuery("SELECT uid from t where uid is NULL order by uid")
- .returns(new Object[] {null})
- .check();
-
- assertThrows("SELECT avg(uid) from t", UnsupportedOperationException.class,
- "AVG() is not supported for type 'UUID'.");
-
- assertThrows("SELECT sum(uid) from t", UnsupportedOperationException.class,
- "SUM() is not supported for type 'UUID'.");
- }
- finally {
- executeSql("DROP TABLE if exists tbl");
- }
+ executeSql("CREATE TABLE t(id INT, name VARCHAR(255), uid UUID, primary key (id)) WITH " + atomicity());
+
+ if (indexed)
+ executeSql("CREATE INDEX uuid_idx ON t (uid);");
+
+ UUID uuid1 = UUID.randomUUID();
+ UUID uuid2 = UUID.randomUUID();
+
+ UUID max = uuid1.compareTo(uuid2) > 0 ? uuid1 : uuid2;
+ UUID min = max == uuid1 ? uuid2 : uuid1;
+
+ executeSql("INSERT INTO t VALUES (1, 'fd10556e-fc27-4a99-b5e4-89b8344cb3ce', '" + uuid1 + "')");
+ // Name == UUID
+ executeSql("INSERT INTO t VALUES (2, '" + uuid2 + "', '" + uuid2 + "')");
+ executeSql("INSERT INTO t VALUES (3, NULL, NULL)");
+
+ assertQuery("SELECT * FROM t")
+ .returns(1, "fd10556e-fc27-4a99-b5e4-89b8344cb3ce", uuid1)
+ .returns(2, uuid2.toString(), uuid2)
+ .returns(3, null, null)
+ .check();
+
+ assertQuery("SELECT uid FROM t WHERE uid < '" + max + "'")
+ .returns(min)
+ .check();
+
+ assertQuery("SELECT uid FROM t WHERE '" + max + "' > uid")
+ .returns(min)
+ .check();
+
+ assertQuery("SELECT * FROM t WHERE name = uid")
+ .returns(2, uuid2.toString(), uuid2)
+ .check();
+
+ assertQuery("SELECT * FROM t WHERE name != uid")
+ .returns(1, "fd10556e-fc27-4a99-b5e4-89b8344cb3ce", uuid1)
+ .check();
+
+ assertQuery("SELECT count(*), uid FROM t group by uid order by uid")
+ .returns(1L, min)
+ .returns(1L, max)
+ .returns(1L, null)
+ .check();
+
+ assertQuery("SELECT count(*), uid FROM t group by uid having uid = " +
+ "'" + uuid1 + "' order by uid")
+ .returns(1L, uuid1)
+ .check();
+
+ assertQuery("SELECT t1.* from t t1, (select * from t) t2 where t1.uid = t2.name")
+ .returns(2, uuid2.toString(), uuid2)
+ .check();
+
+ assertQuery("SELECT t1.* from t t1, (select * from t) t2 " +
+ "where t1.uid = t2.uid")
+ .returns(1, "fd10556e-fc27-4a99-b5e4-89b8344cb3ce", uuid1)
+ .returns(2, uuid2.toString(), uuid2)
+ .check();
+
+ assertQuery("SELECT max(uid) from t")
+ .returns(max)
+ .check();
+
+ assertQuery("SELECT min(uid) from t")
+ .returns(min)
+ .check();
+
+ assertQuery("SELECT uid from t where uid is not NULL order by uid")
+ .returns(min)
+ .returns(max)
+ .check();
+
+ assertQuery("SELECT uid from t where uid is NULL order by uid")
+ .returns(new Object[]{null})
+ .check();
+
+ assertThrows("SELECT avg(uid) from t", UnsupportedOperationException.class,
+ "AVG() is not supported for type 'UUID'.");
+
+ assertThrows("SELECT sum(uid) from t", UnsupportedOperationException.class,
+ "SUM() is not supported for type 'UUID'.");
}
/**
@@ -210,30 +200,25 @@ public class DataTypesTest extends AbstractBasicIntegrationTest {
*/
@Test
public void testNumericRanges() {
- try {
- executeSql("CREATE TABLE tbl(tiny TINYINT, small SMALLINT, i INTEGER, big BIGINT)");
+ executeSql("CREATE TABLE tbl(tiny TINYINT, small SMALLINT, i INTEGER, big BIGINT) WITH " + atomicity());
- executeSql("INSERT INTO tbl VALUES (" + Byte.MAX_VALUE + ", " + Short.MAX_VALUE + ", " +
- Integer.MAX_VALUE + ", " + Long.MAX_VALUE + ')');
+ executeSql("INSERT INTO tbl VALUES (" + Byte.MAX_VALUE + ", " + Short.MAX_VALUE + ", " +
+ Integer.MAX_VALUE + ", " + Long.MAX_VALUE + ')');
- assertQuery("SELECT tiny FROM tbl").returns(Byte.MAX_VALUE).check();
- assertQuery("SELECT small FROM tbl").returns(Short.MAX_VALUE).check();
- assertQuery("SELECT i FROM tbl").returns(Integer.MAX_VALUE).check();
- assertQuery("SELECT big FROM tbl").returns(Long.MAX_VALUE).check();
+ assertQuery("SELECT tiny FROM tbl").returns(Byte.MAX_VALUE).check();
+ assertQuery("SELECT small FROM tbl").returns(Short.MAX_VALUE).check();
+ assertQuery("SELECT i FROM tbl").returns(Integer.MAX_VALUE).check();
+ assertQuery("SELECT big FROM tbl").returns(Long.MAX_VALUE).check();
- executeSql("DELETE from tbl");
+ executeSql("DELETE from tbl");
- executeSql("INSERT INTO tbl VALUES (" + Byte.MIN_VALUE + ", " + Short.MIN_VALUE + ", " +
- Integer.MIN_VALUE + ", " + Long.MIN_VALUE + ')');
+ executeSql("INSERT INTO tbl VALUES (" + Byte.MIN_VALUE + ", " + Short.MIN_VALUE + ", " +
+ Integer.MIN_VALUE + ", " + Long.MIN_VALUE + ')');
- assertQuery("SELECT tiny FROM tbl").returns(Byte.MIN_VALUE).check();
- assertQuery("SELECT small FROM tbl").returns(Short.MIN_VALUE).check();
- assertQuery("SELECT i FROM tbl").returns(Integer.MIN_VALUE).check();
- assertQuery("SELECT big FROM tbl").returns(Long.MIN_VALUE).check();
- }
- finally {
- executeSql("DROP TABLE if exists tbl");
- }
+ assertQuery("SELECT tiny FROM tbl").returns(Byte.MIN_VALUE).check();
+ assertQuery("SELECT small FROM tbl").returns(Short.MIN_VALUE).check();
+ assertQuery("SELECT i FROM tbl").returns(Integer.MIN_VALUE).check();
+ assertQuery("SELECT big FROM tbl").returns(Long.MIN_VALUE).check();
}
/**
@@ -241,38 +226,33 @@ public class DataTypesTest extends AbstractBasicIntegrationTest {
*/
@Test
public void testNumericConvertationOnEquals() {
- try {
- executeSql("CREATE TABLE tbl(tiny TINYINT, small SMALLINT, i INTEGER, big BIGINT)");
+ executeSql("CREATE TABLE tbl(tiny TINYINT, small SMALLINT, i INTEGER, big BIGINT) WITH " + atomicity());
- executeSql("INSERT INTO tbl VALUES (1, 2, 3, 4), (5, 5, 5, 5)");
+ executeSql("INSERT INTO tbl VALUES (1, 2, 3, 4), (5, 5, 5, 5)");
- assertQuery("SELECT t1.tiny FROM tbl t1 JOIN tbl t2 ON (t1.tiny=t2.small)").returns((byte)5).check();
- assertQuery("SELECT t1.small FROM tbl t1 JOIN tbl t2 ON (t1.small=t2.tiny)").returns((short)5).check();
+ assertQuery("SELECT t1.tiny FROM tbl t1 JOIN tbl t2 ON (t1.tiny=t2.small)").returns((byte)5).check();
+ assertQuery("SELECT t1.small FROM tbl t1 JOIN tbl t2 ON (t1.small=t2.tiny)").returns((short)5).check();
- assertQuery("SELECT t1.tiny FROM tbl t1 JOIN tbl t2 ON (t1.tiny=t2.i)").returns((byte)5).check();
- assertQuery("SELECT t1.i FROM tbl t1 JOIN tbl t2 ON (t1.i=t2.tiny)").returns(5).check();
+ assertQuery("SELECT t1.tiny FROM tbl t1 JOIN tbl t2 ON (t1.tiny=t2.i)").returns((byte)5).check();
+ assertQuery("SELECT t1.i FROM tbl t1 JOIN tbl t2 ON (t1.i=t2.tiny)").returns(5).check();
- assertQuery("SELECT t1.tiny FROM tbl t1 JOIN tbl t2 ON (t1.tiny=t2.big)").returns((byte)5).check();
- assertQuery("SELECT t1.big FROM tbl t1 JOIN tbl t2 ON (t1.big=t2.tiny)").returns(5L).check();
+ assertQuery("SELECT t1.tiny FROM tbl t1 JOIN tbl t2 ON (t1.tiny=t2.big)").returns((byte)5).check();
+ assertQuery("SELECT t1.big FROM tbl t1 JOIN tbl t2 ON (t1.big=t2.tiny)").returns(5L).check();
- assertQuery("SELECT t1.small FROM tbl t1 JOIN tbl t2 ON (t1.small=t2.i)").returns((short)5).check();
- assertQuery("SELECT t1.i FROM tbl t1 JOIN tbl t2 ON (t1.i=t2.small)").returns(5).check();
+ assertQuery("SELECT t1.small FROM tbl t1 JOIN tbl t2 ON (t1.small=t2.i)").returns((short)5).check();
+ assertQuery("SELECT t1.i FROM tbl t1 JOIN tbl t2 ON (t1.i=t2.small)").returns(5).check();
- assertQuery("SELECT t1.small FROM tbl t1 JOIN tbl t2 ON (t1.small=t2.big)").returns((short)5).check();
- assertQuery("SELECT t1.big FROM tbl t1 JOIN tbl t2 ON (t1.big=t2.small)").returns(5L).check();
+ assertQuery("SELECT t1.small FROM tbl t1 JOIN tbl t2 ON (t1.small=t2.big)").returns((short)5).check();
+ assertQuery("SELECT t1.big FROM tbl t1 JOIN tbl t2 ON (t1.big=t2.small)").returns(5L).check();
- assertQuery("SELECT t1.i FROM tbl t1 JOIN tbl t2 ON (t1.i=t2.big)").returns(5).check();
- assertQuery("SELECT t1.big FROM tbl t1 JOIN tbl t2 ON (t1.big=t2.i)").returns(5L).check();
- }
- finally {
- executeSql("DROP TABLE if exists tbl");
- }
+ assertQuery("SELECT t1.i FROM tbl t1 JOIN tbl t2 ON (t1.i=t2.big)").returns(5).check();
+ assertQuery("SELECT t1.big FROM tbl t1 JOIN tbl t2 ON (t1.big=t2.i)").returns(5L).check();
}
/** */
@Test
public void testUnicodeStrings() {
- client.getOrCreateCache(new CacheConfiguration<Integer, String>()
+ client.getOrCreateCache(this.<Integer, String>cacheConfiguration()
.setName("string_cache")
.setSqlSchema("PUBLIC")
.setQueryEntities(F.asList(new QueryEntity(Integer.class, String.class).setTableName("string_table")))
@@ -316,109 +296,100 @@ public class DataTypesTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testBinarySql() {
- try {
- executeSql("CREATE TABLE tbl(b BINARY(3), v VARBINARY)");
+ executeSql("CREATE TABLE tbl(b BINARY(3), v VARBINARY) WITH " + atomicity());
- byte[] val = new byte[] {1, 2, 3};
+ byte[] val = new byte[]{1, 2, 3};
- // From parameters to internal, from internal to store, from store to internal and from internal to user.
- executeSql("INSERT INTO tbl VALUES (?, ?)", val, val);
+ // From parameters to internal, from internal to store, from store to internal and from internal to user.
+ executeSql("INSERT INTO tbl VALUES (?, ?)", val, val);
- List<List<?>> res = executeSql("SELECT b, v FROM tbl");
+ List<List<?>> res = executeSql("SELECT b, v FROM tbl");
- assertEquals(1, res.size());
- assertEquals(2, res.get(0).size());
- assertTrue(Objects.deepEquals(val, res.get(0).get(0)));
- assertTrue(Objects.deepEquals(val, res.get(0).get(1)));
+ assertEquals(1, res.size());
+ assertEquals(2, res.get(0).size());
+ assertTrue(Objects.deepEquals(val, res.get(0).get(0)));
+ assertTrue(Objects.deepEquals(val, res.get(0).get(1)));
- executeSql("DELETE FROM tbl");
+ executeSql("DELETE FROM tbl");
- // From literal to internal, from internal to store, from store to internal and from internal to user.
- executeSql("INSERT INTO tbl VALUES (x'1A2B3C', x'AABBCC')");
+ // From literal to internal, from internal to store, from store to internal and from internal to user.
+ executeSql("INSERT INTO tbl VALUES (x'1A2B3C', x'AABBCC')");
- res = executeSql("SELECT b, v FROM tbl");
+ res = executeSql("SELECT b, v FROM tbl");
- assertEquals(1, res.size());
- assertEquals(2, res.get(0).size());
- assertTrue(Objects.deepEquals(new byte[] {0x1A, 0x2B, 0x3C}, res.get(0).get(0)));
- assertTrue(Objects.deepEquals(new byte[] {(byte)0xAA, (byte)0xBB, (byte)0xCC}, res.get(0).get(1)));
- }
- finally {
- executeSql("DROP TABLE IF EXISTS tbl");
- }
+ assertEquals(1, res.size());
+ assertEquals(2, res.get(0).size());
+ assertTrue(Objects.deepEquals(new byte[]{0x1A, 0x2B, 0x3C}, res.get(0).get(0)));
+ assertTrue(Objects.deepEquals(new byte[]{(byte)0xAA, (byte)0xBB, (byte)0xCC}, res.get(0).get(1)));
}
/** Cache API - SQL API cross check. */
@Test
public void testBinaryCache() {
- try {
- IgniteCache<Integer, byte[]> cache = client.getOrCreateCache(new CacheConfiguration<Integer, byte[]>()
- .setName("binary_cache")
- .setSqlSchema("PUBLIC")
- .setQueryEntities(F.asList(new QueryEntity(Integer.class, byte[].class).setTableName("binary_table")))
- );
+ IgniteCache<Integer, byte[]> cache = client.getOrCreateCache(this.<Integer, byte[]>cacheConfiguration()
+ .setName("binary_cache")
+ .setSqlSchema("PUBLIC")
+ .setQueryEntities(F.asList(new QueryEntity(Integer.class, byte[].class).setTableName("binary_table")))
+ );
- byte[] val = new byte[] {1, 2, 3};
+ byte[] val = new byte[]{1, 2, 3};
- // From cache to internal, from internal to user.
- cache.put(1, val);
+ // From cache to internal, from internal to user.
+ put(client, cache, 1, val);
- List<List<?>> res = executeSql("SELECT _val FROM binary_table");
- assertEquals(1, res.size());
- assertEquals(1, res.get(0).size());
- assertTrue(Objects.deepEquals(val, res.get(0).get(0)));
+ List<List<?>> res = executeSql("SELECT _val FROM binary_table");
+ assertEquals(1, res.size());
+ assertEquals(1, res.get(0).size());
+ assertTrue(Objects.deepEquals(val, res.get(0).get(0)));
- // From literal to internal, from internal to cache.
- executeSql("INSERT INTO binary_table (_KEY, _VAL) VALUES (2, x'010203')");
+ // From literal to internal, from internal to cache.
+ executeSql("INSERT INTO binary_table (_KEY, _VAL) VALUES (2, x'010203')");
+
+ SupplierX<Object> check = () -> {
byte[] resVal = cache.get(2);
+
assertTrue(Objects.deepEquals(val, resVal));
- }
- finally {
- client.destroyCache("binary_cache");
- }
+
+ return null;
+ };
+
+ if (sqlTxMode == SqlTransactionMode.NONE)
+ check.get();
+ else
+ txAction(client, check);
}
/** */
@Test
public void testBinaryAggregation() {
- try {
- executeSql("CREATE TABLE tbl(b varbinary)");
- executeSql("INSERT INTO tbl VALUES (NULL)");
- executeSql("INSERT INTO tbl VALUES (x'010203')");
- executeSql("INSERT INTO tbl VALUES (x'040506')");
- List<List<?>> res = executeSql("SELECT MIN(b), MAX(b) FROM tbl");
-
- assertEquals(1, res.size());
- assertEquals(2, res.get(0).size());
- assertTrue(Objects.deepEquals(new byte[] {1, 2, 3}, res.get(0).get(0)));
- assertTrue(Objects.deepEquals(new byte[] {4, 5, 6}, res.get(0).get(1)));
- }
- finally {
- executeSql("DROP TABLE IF EXISTS tbl");
- }
+ executeSql("CREATE TABLE tbl(b varbinary) WITH " + atomicity());
+ executeSql("INSERT INTO tbl VALUES (NULL)");
+ executeSql("INSERT INTO tbl VALUES (x'010203')");
+ executeSql("INSERT INTO tbl VALUES (x'040506')");
+ List<List<?>> res = executeSql("SELECT MIN(b), MAX(b) FROM tbl");
+
+ assertEquals(1, res.size());
+ assertEquals(2, res.get(0).size());
+ assertTrue(Objects.deepEquals(new byte[]{1, 2, 3}, res.get(0).get(0)));
+ assertTrue(Objects.deepEquals(new byte[]{4, 5, 6}, res.get(0).get(1)));
}
/** */
@Test
public void testBinaryConcat() {
- try {
- executeSql("CREATE TABLE tbl(b varbinary)");
- executeSql("INSERT INTO tbl VALUES (x'010203')");
- List<List<?>> res = executeSql("SELECT b || x'040506' FROM tbl");
-
- assertEquals(1, res.size());
- assertEquals(1, res.get(0).size());
- assertTrue(Objects.deepEquals(new byte[] {1, 2, 3, 4, 5, 6}, res.get(0).get(0)));
- }
- finally {
- executeSql("DROP TABLE IF EXISTS tbl");
- }
+ executeSql("CREATE TABLE tbl(b varbinary) WITH " + atomicity());
+ executeSql("INSERT INTO tbl VALUES (x'010203')");
+ List<List<?>> res = executeSql("SELECT b || x'040506' FROM tbl");
+
+ assertEquals(1, res.size());
+ assertEquals(1, res.get(0).size());
+ assertTrue(Objects.deepEquals(new byte[]{1, 2, 3, 4, 5, 6}, res.get(0).get(0)));
}
/** */
@Test
public void testDecimalScale() {
- sql("CREATE TABLE t (id INT PRIMARY KEY, val1 DECIMAL(5, 3), val2 DECIMAL(3), val3 DECIMAL)");
+ sql("CREATE TABLE t (id INT PRIMARY KEY, val1 DECIMAL(5, 3), val2 DECIMAL(3), val3 DECIMAL) WITH " + atomicity());
// Check literals scale.
sql("INSERT INTO t values (0, 0, 0, 0)");
@@ -453,7 +424,7 @@ public class DataTypesTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testNumericConversion() {
- sql("CREATE TABLE t (v1 TINYINT, v2 SMALLINT, v3 INT, v4 BIGINT, v5 DECIMAL, v6 FLOAT, v7 DOUBLE)");
+ sql("CREATE TABLE t (v1 TINYINT, v2 SMALLINT, v3 INT, v4 BIGINT, v5 DECIMAL, v6 FLOAT, v7 DOUBLE) WITH " + atomicity());
List<Number> params = F.asList((byte)1, (short)2, 3, 4L, BigDecimal.valueOf(5), 6f, 7d);
@@ -474,6 +445,8 @@ public class DataTypesTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testFunctionArgsToNumericImplicitConversion() {
+ assumeTrue("Test use queries that doesn't touch any data. Skip for tx modes", sqlTxMode == SqlTransactionMode.NONE);
+
assertQuery("select decode(?, 0, 0, 1, 1.0)").withParams(0).returns(new BigDecimal("0.0")).check();
assertQuery("select decode(?, 0, 0, 1, 1.0)").withParams(1).returns(new BigDecimal("1.0")).check();
assertQuery("select decode(?, 0, 0, 1, 1.000)").withParams(0).returns(new BigDecimal("0.000")).check();
@@ -547,6 +520,8 @@ public class DataTypesTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testCastDecimalOverflows() {
+ assumeTrue("Test use queries that doesn't touch any data. Skip for tx modes", sqlTxMode == SqlTransactionMode.NONE);
+
// BIGINT
assertQuery("SELECT CAST(9223372036854775807.1 AS BIGINT)").returns(9223372036854775807L).check();
assertQuery("SELECT CAST(9223372036854775807.9 AS BIGINT)").returns(9223372036854775807L).check();
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/DateTimeTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/DateTimeTest.java
index 1335e33aa6e..2bf25ef6a18 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/DateTimeTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/DateTimeTest.java
@@ -31,13 +31,13 @@ import org.junit.Test;
import static java.util.Collections.singletonList;
/** */
-public class DateTimeTest extends AbstractBasicIntegrationTest {
+public class DateTimeTest extends AbstractBasicIntegrationTransactionalTest {
/** */
private static final SimpleDateFormat DATE_FORMAT = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
/** {@inheritDoc} */
- @Override protected void beforeTestsStarted() throws Exception {
- super.beforeTestsStarted();
+ @Override protected void init() throws Exception {
+ super.init();
QueryEntity qryEnt = new QueryEntity();
qryEnt.setKeyFieldName("ID");
@@ -51,22 +51,22 @@ public class DateTimeTest extends AbstractBasicIntegrationTest {
qryEnt.addQueryField("SQLTIMESTAMP", Timestamp.class.getName(), null);
qryEnt.setTableName("datetimetable");
- final CacheConfiguration<Integer, DateTimeEntry> cfg = new CacheConfiguration<>(qryEnt.getTableName());
-
- cfg.setCacheMode(CacheMode.PARTITIONED)
+ final CacheConfiguration<Integer, DateTimeEntry> cfg = this.<Integer, DateTimeEntry>cacheConfiguration()
+ .setName(qryEnt.getTableName())
+ .setCacheMode(CacheMode.PARTITIONED)
.setBackups(1)
.setQueryEntities(singletonList(qryEnt))
.setSqlSchema("PUBLIC");
IgniteCache<Integer, DateTimeEntry> dateTimeCache = client.createCache(cfg);
- dateTimeCache.put(1, new DateTimeEntry(1, javaDate("2020-10-01 12:00:00.000"),
+ put(client, dateTimeCache, 1, new DateTimeEntry(1, javaDate("2020-10-01 12:00:00.000"),
sqlDate("2020-10-01"), sqlTime("12:00:00"), sqlTimestamp("2020-10-01 12:00:00.000")));
- dateTimeCache.put(2, new DateTimeEntry(2, javaDate("2020-12-01 00:10:20.000"),
+ put(client, dateTimeCache, 2, new DateTimeEntry(2, javaDate("2020-12-01 00:10:20.000"),
sqlDate("2020-12-01"), sqlTime("00:10:20"), sqlTimestamp("2020-12-01 00:10:20.000")));
- dateTimeCache.put(3, new DateTimeEntry(3, javaDate("2020-10-20 13:15:00.000"),
+ put(client, dateTimeCache, 3, new DateTimeEntry(3, javaDate("2020-10-20 13:15:00.000"),
sqlDate("2020-10-20"), sqlTime("13:15:00"), sqlTimestamp("2020-10-20 13:15:00.000")));
- dateTimeCache.put(4, new DateTimeEntry(4, javaDate("2020-01-01 22:40:00.000"),
+ put(client, dateTimeCache, 4, new DateTimeEntry(4, javaDate("2020-01-01 22:40:00.000"),
sqlDate("2020-01-01"), sqlTime("22:40:00"), sqlTimestamp("2020-01-01 22:40:00.000")));
}
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/HashSpoolIntegrationTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/HashSpoolIntegrationTest.java
index d1ddc5423bf..605e1175687 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/HashSpoolIntegrationTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/HashSpoolIntegrationTest.java
@@ -23,11 +23,11 @@ import org.junit.Test;
/**
* Hash spool test.
*/
-public class HashSpoolIntegrationTest extends AbstractBasicIntegrationTest {
+public class HashSpoolIntegrationTest extends AbstractBasicIntegrationTransactionalTest {
/** */
@Test
public void testNullsInSearchRow() {
- executeSql("CREATE TABLE t(i1 INTEGER, i2 INTEGER)");
+ executeSql("CREATE TABLE t(i1 INTEGER, i2 INTEGER) WITH " + atomicity());
executeSql("INSERT INTO t VALUES (null, 0), (1, 1), (2, 2), (3, null)");
assertQuery("SELECT i1, (SELECT i2 FROM t WHERE i1=t1.i1) FROM t t1")
@@ -50,8 +50,8 @@ public class HashSpoolIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testNullsInSearchRowMultipleColumns() {
- executeSql("CREATE TABLE t0(i1 INTEGER, i2 INTEGER)");
- executeSql("CREATE TABLE t1(i1 INTEGER, i2 INTEGER)");
+ executeSql("CREATE TABLE t0(i1 INTEGER, i2 INTEGER) WITH " + atomicity());
+ executeSql("CREATE TABLE t1(i1 INTEGER, i2 INTEGER) WITH " + atomicity());
executeSql("INSERT INTO t0 VALUES (null, 0), (1, null), (null, 2), (3, null), (1, 1)");
executeSql("INSERT INTO t1 VALUES (null, 0), (null, 1), (2, null), (3, null), (1, 1)");
@@ -67,7 +67,7 @@ public class HashSpoolIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testHashSpoolCondition() {
- executeSql("CREATE TABLE t(i INTEGER)");
+ executeSql("CREATE TABLE t(i INTEGER) WITH " + atomicity());
executeSql("INSERT INTO t VALUES (0), (1), (2)");
String sql = "SELECT i, (SELECT i FROM t WHERE i=t1.i AND i-1=0) FROM t AS t1";
@@ -83,10 +83,10 @@ public class HashSpoolIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testIsNotDistinctFrom() {
- executeSql("CREATE TABLE t1(i1 INTEGER, i2 INTEGER)");
- executeSql("INSERT INTO t1 VALUES (1, null), (2, 2), (null, 3), (3, null)");
+ executeSql("CREATE TABLE t1(i1 INTEGER, i2 INTEGER) WITH " + atomicity());
+ executeSql("CREATE TABLE t2(i3 INTEGER, i4 INTEGER) WITH " + atomicity());
- executeSql("CREATE TABLE t2(i3 INTEGER, i4 INTEGER)");
+ executeSql("INSERT INTO t1 VALUES (1, null), (2, 2), (null, 3), (3, null)");
executeSql("INSERT INTO t2 VALUES (1, 1), (2, 2), (null, 3), (4, null)");
String sql = "SELECT /*+ CNL_JOIN */ i1, i4 FROM t1 JOIN t2 ON i1 IS NOT DISTINCT FROM i3";
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/IndexMultiRangeScanIntegrationTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/IndexMultiRangeScanIntegrationTest.java
index e59be725daf..a4239b8f33a 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/IndexMultiRangeScanIntegrationTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/IndexMultiRangeScanIntegrationTest.java
@@ -16,7 +16,9 @@
*/
package org.apache.ignite.internal.processors.query.calcite.integration;
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.List;
import org.apache.ignite.internal.processors.query.calcite.QueryChecker;
import org.apache.ignite.internal.util.typedef.F;
@@ -29,26 +31,34 @@ import org.junit.runners.Parameterized;
* Tests index multi-range scans (with SEARCH/SARG operator or with dynamic parameters).
*/
@RunWith(Parameterized.class)
-public class IndexMultiRangeScanIntegrationTest extends AbstractBasicIntegrationTest {
+public class IndexMultiRangeScanIntegrationTest extends AbstractBasicIntegrationTransactionalTest {
/** */
- @Parameterized.Parameter
+ @Parameterized.Parameter(1)
public boolean dynamicParams;
- /** */
- @Parameterized.Parameters(name = "dynamicParams={0}")
- public static List<Boolean> parameters() {
- return F.asList(true, false);
+ /** @return Test parameters. */
+ @Parameterized.Parameters(name = "sqlTxMode={0},dynamicParams={1}")
+ public static Collection<?> parameters() {
+ List<Object[]> params = new ArrayList<>();
+
+ for (boolean dynamicParams : new boolean[]{true, false}) {
+ for (SqlTransactionMode sqlTxMode : SqlTransactionMode.values()) {
+ params.add(new Object[]{sqlTxMode, dynamicParams});
+ }
+ }
+
+ return params;
}
/** {@inheritDoc} */
- @Override protected void beforeTestsStarted() throws Exception {
- super.beforeTestsStarted();
+ @Override protected void init() throws Exception {
+ super.init();
- sql("CREATE TABLE test (c1 INTEGER, c2 VARCHAR, c3 INTEGER)");
+ sql("CREATE TABLE test (c1 INTEGER, c2 VARCHAR, c3 INTEGER) WITH " + atomicity());
sql("CREATE INDEX c1c2c3 ON test(c1, c2, c3)");
- sql("CREATE TABLE test_desc (c1 INTEGER, c2 VARCHAR, c3 INTEGER)");
+ sql("CREATE TABLE test_desc (c1 INTEGER, c2 VARCHAR, c3 INTEGER) WITH " + atomicity());
sql("CREATE INDEX c1c2c3_desc ON test_desc(c1 DESC, c2 DESC, c3 DESC)");
- sql("CREATE TABLE test_pk (c1 INTEGER, c2 VARCHAR, c3 INTEGER, PRIMARY KEY(c1, c2, c3))");
+ sql("CREATE TABLE test_pk (c1 INTEGER, c2 VARCHAR, c3 INTEGER, PRIMARY KEY(c1, c2, c3)) WITH " + atomicity());
for (String tbl : F.asList("test", "test_desc", "test_pk")) {
sql("INSERT INTO " + tbl + "(c1, c2, c3) VALUES (0, null, 0)");
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/IndexScanlIntegrationTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/IndexScanlIntegrationTest.java
index 7e7c0df7cbe..df79c0490c0 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/IndexScanlIntegrationTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/IndexScanlIntegrationTest.java
@@ -31,7 +31,6 @@ import org.apache.calcite.util.ImmutableBitSet;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.cache.QueryEntity;
-import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.processors.query.calcite.QueryChecker;
import org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
@@ -51,7 +50,7 @@ import org.junit.Test;
/**
* Index scan test.
*/
-public class IndexScanlIntegrationTest extends AbstractBasicIntegrationTest {
+public class IndexScanlIntegrationTest extends AbstractBasicIntegrationTransactionalTest {
/** */
private static final int ROWS_CNT = 100;
@@ -63,7 +62,7 @@ public class IndexScanlIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testNullsInCNLJSearchRow() {
- executeSql("CREATE TABLE t(i1 INTEGER, i2 INTEGER) WITH TEMPLATE=REPLICATED ");
+ executeSql("CREATE TABLE t(i1 INTEGER, i2 INTEGER) WITH TEMPLATE=REPLICATED," + atomicity());
executeSql("INSERT INTO t VALUES (0, null), (1, null), (2, 2), (3, null), (4, null), (null, 5)");
executeSql("CREATE INDEX t_idx ON t(i1)");
@@ -90,7 +89,7 @@ public class IndexScanlIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testNullsInSearchRow() {
- executeSql("CREATE TABLE t(i1 INTEGER, i2 INTEGER) WITH TEMPLATE=REPLICATED ");
+ executeSql("CREATE TABLE t(i1 INTEGER, i2 INTEGER) WITH TEMPLATE=REPLICATED," + atomicity());
executeSql("INSERT INTO t VALUES (null, 0), (1, null), (2, 2), (3, null)");
executeSql("CREATE INDEX t_idx ON t(i1, i2)");
@@ -127,7 +126,8 @@ public class IndexScanlIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testSegmentedIndexes() {
- IgniteCache<Integer, Employer> emp = client.getOrCreateCache(new CacheConfiguration<Integer, Employer>("emp")
+ IgniteCache<Integer, Employer> emp = client.getOrCreateCache(this.<Integer, Employer>cacheConfiguration()
+ .setName("emp")
.setSqlSchema("PUBLIC")
.setQueryEntities(F.asList(new QueryEntity(Integer.class, Employer.class).setTableName("emp")))
.setQueryParallelism(10)
@@ -165,7 +165,7 @@ public class IndexScanlIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testScanBooleanField() {
- executeSql("CREATE TABLE t(i INTEGER, b BOOLEAN)");
+ executeSql("CREATE TABLE t(i INTEGER, b BOOLEAN) WITH " + atomicity());
executeSql("INSERT INTO t VALUES (0, TRUE), (1, TRUE), (2, FALSE), (3, FALSE), (4, null)");
executeSql("CREATE INDEX t_idx ON t(b)");
@@ -218,11 +218,12 @@ public class IndexScanlIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testIsNotDistinctFrom() {
- executeSql("CREATE TABLE t1(i1 INTEGER) WITH TEMPLATE=REPLICATED");
- executeSql("INSERT INTO t1 VALUES (1), (2), (null), (3)");
+ executeSql("CREATE TABLE t1(i1 INTEGER) WITH TEMPLATE=REPLICATED," + atomicity());
+ executeSql("CREATE TABLE t2(i2 INTEGER, i3 INTEGER) WITH TEMPLATE=REPLICATED," + atomicity());
- executeSql("CREATE TABLE t2(i2 INTEGER, i3 INTEGER) WITH TEMPLATE=REPLICATED");
+ executeSql("INSERT INTO t1 VALUES (1), (2), (null), (3)");
executeSql("INSERT INTO t2 VALUES (1, 1), (2, 2), (null, 3), (4, null)");
+
executeSql("CREATE INDEX t2_idx ON t2(i2)");
String sql = "SELECT /*+ CNL_JOIN */ i1, i3 FROM t1 JOIN t2 ON i1 IS NOT DISTINCT FROM i2";
@@ -262,7 +263,7 @@ public class IndexScanlIntegrationTest extends AbstractBasicIntegrationTest {
checkSingleColumnInlineScan(false, "DECIMAL", BigDecimal::valueOf);
// Multi columns scans.
- executeSql("CREATE TABLE t(id INTEGER PRIMARY KEY, i1 INTEGER, i2 INTEGER, i3 INTEGER)");
+ executeSql("CREATE TABLE t(id INTEGER PRIMARY KEY, i1 INTEGER, i2 INTEGER, i3 INTEGER) WITH " + atomicity());
executeSql("CREATE INDEX t_idx ON t(i1, i3)");
RowCountingIndex idx = injectRowCountingIndex(grid(0), "T", "T_IDX");
@@ -277,7 +278,7 @@ public class IndexScanlIntegrationTest extends AbstractBasicIntegrationTest {
/** */
public void checkSingleColumnInlineScan(boolean expInline, String dataType, IntFunction<Object> valFactory) {
- executeSql("CREATE TABLE t(id INTEGER PRIMARY KEY, val " + dataType + ')');
+ executeSql("CREATE TABLE t(id INTEGER PRIMARY KEY, val " + dataType + ") WITH " + atomicity());
try {
executeSql("CREATE INDEX t_idx ON t(val)");
@@ -294,8 +295,14 @@ public class IndexScanlIntegrationTest extends AbstractBasicIntegrationTest {
if (expInline) {
checker.matches(QueryChecker.containsIndexScan("PUBLIC", "T", "T_IDX")).check();
- assertEquals(ROWS_CNT, idx.rowsProcessed());
- assertTrue(idx.isInlineScan());
+ if (sqlTxMode == SqlTransactionMode.NONE) {
+ assertEquals(ROWS_CNT, idx.rowsProcessed());
+ assertTrue(idx.isInlineScan());
+ }
+ else if (sqlTxMode == SqlTransactionMode.ALL) {
+ assertEquals(0, idx.rowsProcessed());
+ assertFalse(idx.isInlineScan());
+ }
}
else {
checker.check();
@@ -304,6 +311,7 @@ public class IndexScanlIntegrationTest extends AbstractBasicIntegrationTest {
}
}
finally {
+ clearTransaction();
executeSql("DROP TABLE t");
}
}
@@ -323,8 +331,10 @@ public class IndexScanlIntegrationTest extends AbstractBasicIntegrationTest {
if (expInline) {
checker.matches(QueryChecker.containsIndexScan("PUBLIC", "T", "T_IDX")).check();
- assertEquals(ROWS_CNT, idx.rowsProcessed());
- assertTrue(idx.isInlineScan());
+ if (sqlTxMode == SqlTransactionMode.NONE) {
+ assertEquals(ROWS_CNT, idx.rowsProcessed());
+ assertTrue(idx.isInlineScan());
+ }
}
else {
checker.check();
@@ -336,11 +346,12 @@ public class IndexScanlIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testNoIndexHint() {
- executeSql("CREATE TABLE t1(i1 INTEGER) WITH TEMPLATE=PARTITIONED");
- executeSql("INSERT INTO t1 VALUES (1), (2), (30), (40)");
+ executeSql("CREATE TABLE t1(i1 INTEGER) WITH TEMPLATE=PARTITIONED," + atomicity());
executeSql("CREATE INDEX t1_idx ON t1(i1)");
- executeSql("CREATE TABLE t2(i2 INTEGER, i3 INTEGER) WITH TEMPLATE=PARTITIONED");
+ executeSql("CREATE TABLE t2(i2 INTEGER, i3 INTEGER) WITH TEMPLATE=PARTITIONED," + atomicity());
+
+ executeSql("INSERT INTO t1 VALUES (1), (2), (30), (40)");
for (int i = 0; i < 100; ++i)
executeSql("INSERT INTO t2 VALUES (?, ?)", i, i);
@@ -370,12 +381,18 @@ public class IndexScanlIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testForcedIndexHint() {
- executeSql("CREATE TABLE t1(i1 INTEGER, i2 INTEGER, i3 INTEGER) WITH TEMPLATE=PARTITIONED");
+ executeSql("CREATE TABLE t1(i1 INTEGER, i2 INTEGER, i3 INTEGER) WITH TEMPLATE=PARTITIONED," + atomicity());
executeSql("CREATE INDEX t1_idx1 ON t1(i1)");
executeSql("CREATE INDEX t1_idx2 ON t1(i2)");
executeSql("CREATE INDEX t1_idx3 ON t1(i3)");
+ executeSql("CREATE TABLE t2(i21 INTEGER, i22 INTEGER, i23 INTEGER) WITH TEMPLATE=PARTITIONED," + atomicity());
+
+ executeSql("CREATE INDEX t2_idx1 ON t2(i21)");
+ executeSql("CREATE INDEX t2_idx2 ON t2(i22)");
+ executeSql("CREATE INDEX t2_idx3 ON t2(i23)");
+
executeSql("INSERT INTO t1 VALUES (1, 2, 3)");
assertQuery("SELECT /*+ FORCE_INDEX(T1_IDX1) */ i1 FROM t1 where i1=1 and i2=2 and i3=3")
@@ -393,15 +410,9 @@ public class IndexScanlIntegrationTest extends AbstractBasicIntegrationTest {
.returns(1)
.check();
- executeSql("CREATE TABLE t2(i21 INTEGER, i22 INTEGER, i23 INTEGER) WITH TEMPLATE=PARTITIONED");
-
for (int i = 99; i < 300; ++i)
executeSql("INSERT INTO t2 VALUES (?, ?, ?)", i + 1, i + 1, i + 1);
- executeSql("CREATE INDEX t2_idx1 ON t2(i21)");
- executeSql("CREATE INDEX t2_idx2 ON t2(i22)");
- executeSql("CREATE INDEX t2_idx3 ON t2(i23)");
-
assertQuery("SELECT /*+ FORCE_INDEX(T1_IDX2), FORCE_INDEX(T2_IDX2) */ i1, i22 FROM t1, t2 where i2=i22 " +
"and i3=i23 + 1")
.matches(QueryChecker.containsIndexScan("PUBLIC", "T1", "T1_IDX2"))
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/JoinIntegrationTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/JoinIntegrationTest.java
index d41bb01eb58..603a7123b36 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/JoinIntegrationTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/JoinIntegrationTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.ignite.internal.processors.query.calcite.integration;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;
@@ -27,25 +28,31 @@ import org.junit.runners.Parameterized;
/** */
@RunWith(Parameterized.class)
-public class JoinIntegrationTest extends AbstractBasicIntegrationTest {
+public class JoinIntegrationTest extends AbstractBasicIntegrationTransactionalTest {
/** */
- @Parameterized.Parameter
+ @Parameterized.Parameter(1)
public JoinType joinType;
/** */
- @Parameterized.Parameters(name = "joinType={0}")
+ @Parameterized.Parameters(name = "sqlTxMode={0},joinType={1}")
public static List<Object[]> params() {
- return Arrays.stream(JoinType.values())
- .map(jt -> new Object[]{jt})
- .collect(Collectors.toList());
+ List<Object[]> params = new ArrayList<>();
+
+ for (SqlTransactionMode sqlTxMode : SqlTransactionMode.values()) {
+ for (JoinType jt : JoinType.values()) {
+ params.add(new Object[]{sqlTxMode, jt});
+ }
+ }
+
+ return params;
}
/** {@inheritDoc} */
- @Override protected void beforeTestsStarted() throws Exception {
- super.beforeTestsStarted();
+ @Override protected void init() throws Exception {
+ super.init();
- executeSql("create table t1 (c1 int, c2 int, c3 int)");
- executeSql("create table t2 (c1 int, c2 int, c3 int)");
+ executeSql("create table t1 (c1 int, c2 int, c3 int) WITH " + atomicity());
+ executeSql("create table t2 (c1 int, c2 int, c3 int) WITH " + atomicity());
executeSql("create index t1_idx on t1 (c3, c2, c1)");
executeSql("create index t2_idx on t2 (c3, c2, c1)");
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/JoinRehashIntegrationTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/JoinRehashIntegrationTest.java
index 8b6c46aba79..2370eea664f 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/JoinRehashIntegrationTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/JoinRehashIntegrationTest.java
@@ -23,7 +23,7 @@ import org.apache.ignite.testframework.GridTestUtils;
import org.junit.Test;
/** */
-public class JoinRehashIntegrationTest extends AbstractBasicIntegrationTest {
+public class JoinRehashIntegrationTest extends AbstractBasicIntegrationTransactionalTest {
/** {@inheritDoc} */
@Override protected int nodeCount() {
return 3;
@@ -81,13 +81,13 @@ public class JoinRehashIntegrationTest extends AbstractBasicIntegrationTest {
" price decimal,\n" +
" amount int,\n" +
" PRIMARY KEY (id))\n" +
- " WITH \"cache_name=order_items,backups=1\"");
+ " WITH \"cache_name=order_items,backups=1," + atomicity() + "\"");
sql("CREATE TABLE orders (\n" +
" id int,\n" +
" region varchar,\n" +
" PRIMARY KEY (id))\n" +
- " WITH \"cache_name=orders,backups=1\"");
+ " WITH \"cache_name=orders,backups=1," + atomicity() + "\"");
sql("CREATE INDEX order_items_orderId ON order_items (orderId ASC)");
sql("CREATE INDEX orders_region ON orders (region ASC)");
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/KeepBinaryIntegrationTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/KeepBinaryIntegrationTest.java
index f3aff6c99c2..bf62dd6b833 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/KeepBinaryIntegrationTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/KeepBinaryIntegrationTest.java
@@ -25,7 +25,6 @@ import org.apache.ignite.binary.BinaryObject;
import org.apache.ignite.cache.query.SqlFieldsQuery;
import org.apache.ignite.cache.query.annotations.QuerySqlField;
import org.apache.ignite.calcite.CalciteQueryEngineConfiguration;
-import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.util.typedef.F;
import org.junit.Test;
@@ -33,7 +32,7 @@ import org.junit.Test;
/**
* Test "keep binary" in cache queries.
*/
-public class KeepBinaryIntegrationTest extends AbstractBasicIntegrationTest {
+public class KeepBinaryIntegrationTest extends AbstractBasicIntegrationTransactionalTest {
/** */
private static final String CACHE_NAME = "cache";
@@ -41,7 +40,7 @@ public class KeepBinaryIntegrationTest extends AbstractBasicIntegrationTest {
@Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
cfg.getSqlConfiguration().setQueryEnginesConfiguration(new CalciteQueryEngineConfiguration());
- cfg.setCacheConfiguration(new CacheConfiguration<>(CACHE_NAME).setIndexedTypes(Integer.class, Person.class));
+ cfg.setCacheConfiguration(cacheConfiguration().setName(CACHE_NAME).setIndexedTypes(Integer.class, Person.class));
return cfg;
}
@@ -58,35 +57,44 @@ public class KeepBinaryIntegrationTest extends AbstractBasicIntegrationTest {
Person p2 = new Person(2, "name2", F.asList(new Person(3, "name3", null),
new Person(4, "name4", null)));
- cache.put(0, p0);
- cache.put(1, p1);
- cache.put(2, p2);
-
- List<List<?>> res = cache.query(new SqlFieldsQuery("SELECT _VAL, obj, name FROM Person ORDER BY id")).getAll();
-
- assertEquals(3, res.size());
- assertEquals(p0, res.get(0).get(0));
- assertEquals(p1, res.get(1).get(0));
- assertEquals(p2, res.get(2).get(0));
- assertEquals(p0.obj, res.get(0).get(1));
- assertEquals(p1.obj, res.get(1).get(1));
- assertEquals(p2.obj, res.get(2).get(1));
- assertEquals(p0.name, res.get(0).get(2));
- assertEquals(p1.name, res.get(1).get(2));
- assertEquals(p2.name, res.get(2).get(2));
-
- res = cache.withKeepBinary().query(new SqlFieldsQuery("SELECT _VAL, obj, name FROM Person ORDER BY id")).getAll();
-
- assertEquals(3, res.size());
- assertTrue(res.get(0).get(0) instanceof BinaryObject);
- assertTrue(res.get(1).get(0) instanceof BinaryObject);
- assertTrue(res.get(1).get(0) instanceof BinaryObject);
- assertNull(res.get(0).get(1));
- assertTrue(res.get(1).get(1) instanceof BinaryObject);
- assertTrue(res.get(1).get(1) instanceof BinaryObject);
- assertEquals(p0.name, res.get(0).get(2));
- assertEquals(p1.name, res.get(1).get(2));
- assertEquals(p2.name, res.get(2).get(2));
+ put(client, cache, 0, p0);
+ put(client, cache, 1, p1);
+ put(client, cache, 2, p2);
+
+ SupplierX<?> checker = () -> {
+ List<List<?>> res = cache.query(new SqlFieldsQuery("SELECT _VAL, obj, name FROM Person ORDER BY id")).getAll();
+
+ assertEquals(3, res.size());
+ assertEquals(p0, res.get(0).get(0));
+ assertEquals(p1, res.get(1).get(0));
+ assertEquals(p2, res.get(2).get(0));
+ assertEquals(p0.obj, res.get(0).get(1));
+ assertEquals(p1.obj, res.get(1).get(1));
+ assertEquals(p2.obj, res.get(2).get(1));
+ assertEquals(p0.name, res.get(0).get(2));
+ assertEquals(p1.name, res.get(1).get(2));
+ assertEquals(p2.name, res.get(2).get(2));
+
+ res = cache.withKeepBinary().query(new SqlFieldsQuery("SELECT _VAL, obj, name FROM Person ORDER BY id")).getAll();
+
+ assertEquals(3, res.size());
+ assertTrue(res.get(0).get(0) instanceof BinaryObject);
+ assertTrue(res.get(1).get(0) instanceof BinaryObject);
+ assertTrue(res.get(1).get(0) instanceof BinaryObject);
+ assertNull(res.get(0).get(1));
+ assertTrue(res.get(1).get(1) instanceof BinaryObject);
+ assertTrue(res.get(1).get(1) instanceof BinaryObject);
+ assertEquals(p0.name, res.get(0).get(2));
+ assertEquals(p1.name, res.get(1).get(2));
+ assertEquals(p2.name, res.get(2).get(2));
+
+ return null;
+ };
+
+ if (sqlTxMode == SqlTransactionMode.NONE)
+ checker.get();
+ else
+ txAction(client, checker);
}
/** */
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/LimitOffsetIntegrationTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/LimitOffsetIntegrationTest.java
index 71c40bc818c..89b2d73ffda 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/LimitOffsetIntegrationTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/LimitOffsetIntegrationTest.java
@@ -24,7 +24,6 @@ import org.apache.calcite.sql.validate.SqlValidatorException;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.QueryEntity;
-import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.processors.query.IgniteSQLException;
import org.apache.ignite.internal.processors.query.calcite.exec.rel.AbstractNode;
@@ -38,7 +37,7 @@ import static java.util.Collections.singletonList;
/**
* Limit / offset tests.
*/
-public class LimitOffsetIntegrationTest extends AbstractBasicIntegrationTest {
+public class LimitOffsetIntegrationTest extends AbstractBasicIntegrationTransactionalTest {
/** */
private static IgniteCache<Integer, String> cacheRepl;
@@ -46,7 +45,9 @@ public class LimitOffsetIntegrationTest extends AbstractBasicIntegrationTest {
private static IgniteCache<Integer, String> cachePart;
/** {@inheritDoc} */
- @Override protected void beforeTest() throws Exception {
+ @Override protected void init() throws Exception {
+ super.init();
+
cacheRepl = client.cache("TEST_REPL");
cachePart = client.cache("TEST_PART");
}
@@ -78,11 +79,13 @@ public class LimitOffsetIntegrationTest extends AbstractBasicIntegrationTest {
return super.getConfiguration(igniteInstanceName)
.setCacheConfiguration(
- new CacheConfiguration<>(eRepl.getTableName())
+ cacheConfiguration()
+ .setName(eRepl.getTableName())
.setCacheMode(CacheMode.REPLICATED)
.setQueryEntities(singletonList(eRepl))
.setSqlSchema("PUBLIC"),
- new CacheConfiguration<>(ePart.getTableName())
+ cacheConfiguration()
+ .setName(ePart.getTableName())
.setCacheMode(CacheMode.PARTITIONED)
.setQueryEntities(singletonList(ePart))
.setSqlSchema("PUBLIC"));
@@ -188,10 +191,12 @@ public class LimitOffsetIntegrationTest extends AbstractBasicIntegrationTest {
* @param rows Rows count.
*/
private void fillCache(IgniteCache<Integer, String> c, int rows) throws InterruptedException {
+ clearTransaction();
+
c.clear();
for (int i = 0; i < rows; ++i)
- c.put(i, "val_" + String.format("%05d", i));
+ put(client, c, i, "val_" + String.format("%05d", i));
awaitPartitionMapExchange();
}
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/LocalDateTimeSupportTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/LocalDateTimeSupportTest.java
index 6bda9cf2075..1b1f5bf7d52 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/LocalDateTimeSupportTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/LocalDateTimeSupportTest.java
@@ -31,8 +31,8 @@ import java.util.Date;
import java.util.List;
import java.util.function.Function;
import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
import org.apache.ignite.cache.QueryEntity;
-import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.processors.cacheobject.IgniteCacheObjectProcessor;
@@ -48,130 +48,133 @@ import static org.apache.ignite.internal.cache.query.index.sorted.inline.types.D
/** */
@RunWith(Parameterized.class)
-public class LocalDateTimeSupportTest extends AbstractBasicIntegrationTest {
+public class LocalDateTimeSupportTest extends AbstractBasicIntegrationTransactionalTest {
/** */
- @Parameterized.Parameter()
+ @Parameterized.Parameter(1)
public boolean isValidationEnabled;
/** */
- @Parameterized.Parameter(1)
+ @Parameterized.Parameter(2)
public String sqlType;
/** */
- @Parameterized.Parameter(2)
+ @Parameterized.Parameter(3)
public Class<?> colType;
/** */
- @Parameterized.Parameter(3)
+ @Parameterized.Parameter(4)
public Class<?> objType;
/** */
- @Parameterized.Parameter(4)
+ @Parameterized.Parameter(5)
public Function<Object, Object> sqlTypeConverter;
/** */
- @Parameterized.Parameter(5)
+ @Parameterized.Parameter(6)
public boolean isOldDate;
/** */
- @Parameterized.Parameters(name = "isValidationEnabled={0}, sqlType={1}, columnCls={2}, testObjCls={3}, beforeGregorian={5}")
+ @Parameterized.Parameters(
+ name = "sqlTxMode={0}, isValidationEnabled={1}, sqlType={2}, columnCls={3}, testObjCls={4}, beforeGregorian={6}")
public static Collection<Object[]> parameters() {
Collection<Object[]> params = new ArrayList<>();
- for (boolean isOldDate: Arrays.asList(true, false)) {
- for (boolean isV : Arrays.asList(true, false)) {
- params.add(new Object[] {
- isV, "TIMESTAMP", null, LocalDateTime.class, f(ts -> convertToTimestamp((LocalDateTime)ts)), isOldDate
- });
- params.add(new Object[] {
- isV, "TIMESTAMP", null, Date.class, f(ts -> new Timestamp(((Date)ts).getTime())), isOldDate
- });
- params.add(new Object[] {
- isV, "TIMESTAMP", null, java.sql.Date.class, f(ts -> new Timestamp(((Date)ts).getTime())), isOldDate
- });
-
- for (Class<?> testObjCls : Arrays.asList(Timestamp.class, LocalDateTime.class, Date.class, java.sql.Date.class)) {
+ for (SqlTransactionMode sqlTxmode : SqlTransactionMode.values()) {
+ for (boolean isOldDate: Arrays.asList(true, false)) {
+ for (boolean isV : Arrays.asList(true, false)) {
params.add(new Object[] {
- isV, null, Timestamp.class, testObjCls, f(ts -> {
- if (ts instanceof LocalDateTime)
- return convertToTimestamp((LocalDateTime)ts);
- return ts;
- }),
- isOldDate
+ sqlTxmode, isV, "TIMESTAMP", null, LocalDateTime.class, f(ts -> convertToTimestamp((LocalDateTime)ts)), isOldDate
});
-
params.add(new Object[] {
- isV, null, Date.class, testObjCls, f(ts -> {
- if (testObjCls == LocalDateTime.class)
- return new Date(convertToTimestamp((LocalDateTime)ts).getTime());
- else if (testObjCls == Timestamp.class)
- return new Date(((Timestamp)ts).getTime());
- return ts;
- }),
- isOldDate
+ sqlTxmode, isV, "TIMESTAMP", null, Date.class, f(ts -> new Timestamp(((Date)ts).getTime())), isOldDate
});
-
params.add(new Object[] {
- isV, null, LocalDateTime.class, testObjCls, f(ts -> {
- if (testObjCls == Timestamp.class)
- return ((Timestamp)ts).toLocalDateTime();
- else if (testObjCls == java.util.Date.class)
- return new Timestamp(((Date)ts).getTime()).toLocalDateTime();
- else if (testObjCls == java.sql.Date.class)
- return ((java.sql.Date)ts).toLocalDate().atStartOfDay();
- else
- return ts;
- }),
- isOldDate
+ sqlTxmode, isV, "TIMESTAMP", null, java.sql.Date.class, f(ts -> new Timestamp(((Date)ts).getTime())), isOldDate
});
- }
-
- params.add(new Object[] {
- isV, "DATE", null, LocalDate.class, f(d -> convertToSqlDate((LocalDate)d)), isOldDate
- });
- for (Class<?> testObjCls : Arrays.asList(LocalDate.class, java.sql.Date.class)) {
- params.add(new Object[] {
- isV, null, java.sql.Date.class, testObjCls, f(ts -> {
- if (testObjCls == LocalDate.class)
- return convertToSqlDate((LocalDate)ts);
- return ts;
- }),
- isOldDate
- });
+ for (Class<?> testObjCls : Arrays.asList(Timestamp.class, LocalDateTime.class, Date.class, java.sql.Date.class)) {
+ params.add(new Object[] {
+ sqlTxmode, isV, null, Timestamp.class, testObjCls, f(ts -> {
+ if (ts instanceof LocalDateTime)
+ return convertToTimestamp((LocalDateTime)ts);
+ return ts;
+ }),
+ isOldDate
+ });
+
+ params.add(new Object[] {
+ sqlTxmode, isV, null, Date.class, testObjCls, f(ts -> {
+ if (testObjCls == LocalDateTime.class)
+ return new Date(convertToTimestamp((LocalDateTime)ts).getTime());
+ else if (testObjCls == Timestamp.class)
+ return new Date(((Timestamp)ts).getTime());
+ return ts;
+ }),
+ isOldDate
+ });
+
+ params.add(new Object[] {
+ sqlTxmode, isV, null, LocalDateTime.class, testObjCls, f(ts -> {
+ if (testObjCls == Timestamp.class)
+ return ((Timestamp)ts).toLocalDateTime();
+ else if (testObjCls == java.util.Date.class)
+ return new Timestamp(((Date)ts).getTime()).toLocalDateTime();
+ else if (testObjCls == java.sql.Date.class)
+ return ((java.sql.Date)ts).toLocalDate().atStartOfDay();
+ else
+ return ts;
+ }),
+ isOldDate
+ });
+ }
params.add(new Object[] {
- isV, null, LocalDate.class, testObjCls, f(ts -> {
- if (testObjCls == java.sql.Date.class)
- return ((java.sql.Date)ts).toLocalDate();
- return ts;
- }),
- isOldDate
+ sqlTxmode, isV, "DATE", null, LocalDate.class, f(d -> convertToSqlDate((LocalDate)d)), isOldDate
});
- }
- params.add(new Object[] {
- isV, "TIME", null, LocalTime.class, f(t -> convertToSqlTime((LocalTime)t)), isOldDate
- });
+ for (Class<?> testObjCls : Arrays.asList(LocalDate.class, java.sql.Date.class)) {
+ params.add(new Object[] {
+ sqlTxmode, isV, null, java.sql.Date.class, testObjCls, f(ts -> {
+ if (testObjCls == LocalDate.class)
+ return convertToSqlDate((LocalDate)ts);
+ return ts;
+ }),
+ isOldDate
+ });
+
+ params.add(new Object[] {
+ sqlTxmode, isV, null, LocalDate.class, testObjCls, f(ts -> {
+ if (testObjCls == java.sql.Date.class)
+ return ((java.sql.Date)ts).toLocalDate();
+ return ts;
+ }),
+ isOldDate
+ });
+ }
- for (Class<?> testObjCls : Arrays.asList(LocalTime.class, java.sql.Time.class)) {
params.add(new Object[] {
- isV, null, java.sql.Time.class, testObjCls, f(ts -> {
- if (testObjCls == LocalTime.class)
- return convertToSqlTime((LocalTime)ts);
- return ts;
- }),
- isOldDate
+ sqlTxmode, isV, "TIME", null, LocalTime.class, f(t -> convertToSqlTime((LocalTime)t)), isOldDate
});
- params.add(new Object[] {
- isV, null, LocalTime.class, testObjCls, f(ts -> {
- if (testObjCls == java.sql.Time.class)
- return ((java.sql.Time)ts).toLocalTime();
- return ts;
- }),
- isOldDate
- });
+ for (Class<?> testObjCls : Arrays.asList(LocalTime.class, java.sql.Time.class)) {
+ params.add(new Object[] {
+ sqlTxmode, isV, null, java.sql.Time.class, testObjCls, f(ts -> {
+ if (testObjCls == LocalTime.class)
+ return convertToSqlTime((LocalTime)ts);
+ return ts;
+ }),
+ isOldDate
+ });
+
+ params.add(new Object[] {
+ sqlTxmode, isV, null, LocalTime.class, testObjCls, f(ts -> {
+ if (testObjCls == java.sql.Time.class)
+ return ((java.sql.Time)ts).toLocalTime();
+ return ts;
+ }),
+ isOldDate
+ });
+ }
}
}
}
@@ -210,9 +213,10 @@ public class LocalDateTimeSupportTest extends AbstractBasicIntegrationTest {
executeSql("INSERT INTO DATA(_key, id, data) values(?, ?, ?)", 0, 0, testObj);
- grid(0).cache(DEFAULT_CACHE_NAME).put(1, new Data(1, testObj));
+ IgniteCache<Object, Object> cache = client.cache(DEFAULT_CACHE_NAME);
- grid(0).cache(DEFAULT_CACHE_NAME).put(2, grid(0).binary().toBinary(new Data(2, testObj)));
+ put(client, cache, 1, new Data(1, testObj));
+ put(client, cache, 2, client.binary().toBinary(new Data(2, testObj)));
List<List<?>> selectData = executeSql("SELECT data FROM DATA");
@@ -227,7 +231,8 @@ public class LocalDateTimeSupportTest extends AbstractBasicIntegrationTest {
executeSql("CREATE TABLE DATA (id INT PRIMARY KEY, data " + sqlType + ") WITH" +
" \"KEY_TYPE=java.lang.Integer" +
", VALUE_TYPE=" + Data.class.getName() +
- ", CACHE_NAME=default\"");
+ ", CACHE_NAME=default" +
+ ", " + atomicity() + "\"");
}
else {
QueryEntity projEntity = new QueryEntity();
@@ -238,7 +243,8 @@ public class LocalDateTimeSupportTest extends AbstractBasicIntegrationTest {
projEntity.setTableName("DATA");
- grid(0).createCache(new CacheConfiguration<>(DEFAULT_CACHE_NAME)
+ client.createCache(cacheConfiguration()
+ .setName(DEFAULT_CACHE_NAME)
.setQueryEntities(singletonList(projEntity))
.setSqlSchema("PUBLIC"));
}
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/ServerStatisticsIntegrationTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/ServerStatisticsIntegrationTest.java
index f69673b5ab8..6d1bdbb9296 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/ServerStatisticsIntegrationTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/ServerStatisticsIntegrationTest.java
@@ -27,9 +27,9 @@ import java.util.HashSet;
import java.util.Set;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.QueryEntity;
import org.apache.ignite.cache.query.annotations.QuerySqlField;
-import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.processors.query.QueryEngine;
import org.apache.ignite.internal.processors.query.calcite.CalciteQueryProcessor;
@@ -44,7 +44,7 @@ import org.junit.Test;
/**
* Tests for server side statistics usage.
*/
-public class ServerStatisticsIntegrationTest extends AbstractBasicIntegrationTest {
+public class ServerStatisticsIntegrationTest extends AbstractBasicIntegrationTransactionalTest {
/** Server instance. */
private IgniteEx srv;
@@ -86,10 +86,15 @@ public class ServerStatisticsIntegrationTest extends AbstractBasicIntegrationTes
};
/** {@inheritDoc} */
- @Override protected void beforeTestsStarted() throws Exception {
- super.beforeTestsStarted();
+ @Override protected void init() throws Exception {
+ super.init();
+
+ createAllTypesTable();
+
+ createAndPopulateTable(grid(0), 2, CacheMode.PARTITIONED);
+
+ populateAllTypesTable(0, ROW_COUNT);
- createAndPopulateAllTypesTable(0, ROW_COUNT);
}
/** {@inheritDoc} */
@@ -112,8 +117,6 @@ public class ServerStatisticsIntegrationTest extends AbstractBasicIntegrationTes
public void testQueryCostWithStatistics() throws IgniteCheckedException {
String sql = "select name from person where salary is not null";
- createAndPopulateTable();
-
StatisticsKey key = new StatisticsKey("PUBLIC", "PERSON");
srv = ignite(0);
@@ -421,8 +424,6 @@ public class ServerStatisticsIntegrationTest extends AbstractBasicIntegrationTes
*/
@Test
public void testSizeIntOverflow() {
- createAndPopulateTable();
-
F.first(grid(0).context().cache().cache(TABLE_NAME).context().offheap().cacheDataStores())
.updateSize(CU.cacheId(TABLE_NAME), 1L + Integer.MAX_VALUE);
@@ -449,7 +450,7 @@ public class ServerStatisticsIntegrationTest extends AbstractBasicIntegrationTes
* @throws IgniteCheckedException In case of errors.
*/
protected void collectStatistics(StatisticsKey key) throws IgniteCheckedException {
- executeSql(String.format("ANALYZE %s.%s", key.schema(), key.obj()));
+ executeSql(grid(0), String.format("ANALYZE %s.%s", key.schema(), key.obj()));
assertTrue(
GridTestUtils.waitForCondition(
@@ -465,12 +466,12 @@ public class ServerStatisticsIntegrationTest extends AbstractBasicIntegrationTes
* @param key Statistics key to collect statistics for.
*/
protected void dropStatistics(StatisticsKey key) {
- executeSql(String.format("DROP STATISTICS %s.%s", key.schema(), key.obj()));
+ executeSql(grid(0), String.format("DROP STATISTICS %s.%s", key.schema(), key.obj()));
}
/** */
protected QueryChecker assertQuerySrv(String qry) {
- return new QueryChecker(qry) {
+ return new QueryChecker(qry, tx, sqlTxMode) {
@Override protected QueryEngine getEngine() {
return Commons.lookupComponent(srv.context(), QueryEngine.class);
}
@@ -484,23 +485,28 @@ public class ServerStatisticsIntegrationTest extends AbstractBasicIntegrationTes
* @param count Rows count.
* @return Populated cache.
*/
- protected IgniteCache<Integer, AllTypes> createAndPopulateAllTypesTable(int start, int count) {
- IgniteCache<Integer, AllTypes> all_types = grid(0).getOrCreateCache(new CacheConfiguration<Integer, AllTypes>()
- .setName("all_types")
- .setSqlSchema("PUBLIC")
- .setQueryEntities(F.asList(new QueryEntity(Integer.class, AllTypes.class).setTableName("all_types")))
- .setBackups(2)
- );
+ protected IgniteCache<Integer, AllTypes> populateAllTypesTable(int start, int count) {
+ IgniteCache<Integer, AllTypes> all_types = grid(0).cache("all_types");
for (int i = start; i < start + count; i++) {
boolean null_values = (i & 3) == 1;
- all_types.put(i, new AllTypes(i, null_values));
+ put(grid(0), all_types, i, new AllTypes(i, null_values));
}
return all_types;
}
+ /** */
+ private void createAllTypesTable() {
+ grid(0).getOrCreateCache(this.<Integer, AllTypes>cacheConfiguration()
+ .setName("all_types")
+ .setSqlSchema("PUBLIC")
+ .setQueryEntities(F.asList(new QueryEntity(Integer.class, AllTypes.class).setTableName("all_types")))
+ .setBackups(2)
+ );
+ }
+
/**
* Test class with fields of all types.
*/
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/SetOpIntegrationTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/SetOpIntegrationTest.java
index bc87bbb8e59..584c967f039 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/SetOpIntegrationTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/SetOpIntegrationTest.java
@@ -18,8 +18,6 @@
package org.apache.ignite.internal.processors.query.calcite.integration;
import java.util.List;
-
-import javax.cache.Cache;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteDataStreamer;
import org.apache.ignite.cache.CacheMode;
@@ -32,41 +30,47 @@ import org.junit.Test;
/**
* Integration test for set op (EXCEPT, INTERSECT).
*/
-public class SetOpIntegrationTest extends AbstractBasicIntegrationTest {
+public class SetOpIntegrationTest extends AbstractBasicIntegrationTransactionalTest {
/** */
- private void populateTables() throws InterruptedException {
- IgniteCache<Integer, Employer> emp1 = client.getOrCreateCache(new CacheConfiguration<Integer, Employer>("emp1")
+ private void createTables() {
+ client.getOrCreateCache(this.<Integer, Employer>cacheConfiguration()
+ .setName("emp1")
.setSqlSchema("PUBLIC")
.setQueryEntities(F.asList(new QueryEntity(Integer.class, Employer.class).setTableName("emp1")))
.setBackups(2)
);
- IgniteCache<Integer, Employer> emp2 = client.getOrCreateCache(new CacheConfiguration<Integer, Employer>("emp2")
+ client.getOrCreateCache(this.<Integer, Employer>cacheConfiguration()
+ .setName("emp2")
.setSqlSchema("PUBLIC")
.setQueryEntities(F.asList(new QueryEntity(Integer.class, Employer.class).setTableName("emp2")))
.setBackups(1)
);
+ }
- emp1.put(1, new Employer("Igor", 10d));
- emp1.put(2, new Employer("Igor", 11d));
- emp1.put(3, new Employer("Igor", 12d));
- emp1.put(4, new Employer("Igor1", 13d));
- emp1.put(5, new Employer("Igor1", 13d));
- emp1.put(6, new Employer("Igor1", 13d));
- emp1.put(7, new Employer("Roman", 14d));
-
- emp2.put(1, new Employer("Roman", 10d));
- emp2.put(2, new Employer("Roman", 11d));
- emp2.put(3, new Employer("Roman", 12d));
- emp2.put(4, new Employer("Roman", 13d));
- emp2.put(5, new Employer("Igor1", 13d));
- emp2.put(6, new Employer("Igor1", 13d));
+ /** */
+ private void populateEmp1(IgniteCache<Integer, Employer> emp1) {
+ put(client, emp1, 1, new Employer("Igor", 10d));
+ put(client, emp1, 2, new Employer("Igor", 11d));
+ put(client, emp1, 3, new Employer("Igor", 12d));
+ put(client, emp1, 4, new Employer("Igor1", 13d));
+ put(client, emp1, 5, new Employer("Igor1", 13d));
+ put(client, emp1, 6, new Employer("Igor1", 13d));
+ put(client, emp1, 7, new Employer("Roman", 14d));
+ }
- awaitPartitionMapExchange(true, true, null);
+ /** */
+ private void populateEmp2(IgniteCache<Integer, Employer> emp2) {
+ put(client, emp2, 1, new Employer("Roman", 10d));
+ put(client, emp2, 2, new Employer("Roman", 11d));
+ put(client, emp2, 3, new Employer("Roman", 12d));
+ put(client, emp2, 4, new Employer("Roman", 13d));
+ put(client, emp2, 5, new Employer("Igor1", 13d));
+ put(client, emp2, 6, new Employer("Igor1", 13d));
}
- /** Copy cache with it's content to new replicated cache. */
- private void copyCacheAsReplicated(String cacheName) throws InterruptedException {
+ /** */
+ private void createCacheCopyAsReplicated(String cacheName) throws InterruptedException {
IgniteCache<Object, Object> cache = client.cache(cacheName);
CacheConfiguration<Object, Object> ccfg = new CacheConfiguration<Object, Object>(
@@ -76,10 +80,7 @@ public class SetOpIntegrationTest extends AbstractBasicIntegrationTest {
ccfg.setCacheMode(CacheMode.REPLICATED);
ccfg.getQueryEntities().forEach(qe -> qe.setTableName(qe.getTableName() + "_repl"));
- IgniteCache<Object, Object> replCache = client.getOrCreateCache(ccfg);
-
- for (Cache.Entry<?, ?> entry : cache)
- replCache.put(entry.getKey(), entry.getValue());
+ client.getOrCreateCache(ccfg);
awaitPartitionMapExchange(true, true, null);
}
@@ -87,7 +88,9 @@ public class SetOpIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testExcept() throws Exception {
- populateTables();
+ createTables();
+ populateEmp1(client.cache("emp1"));
+ populateEmp2(client.cache("emp2"));
List<List<?>> rows = executeSql("SELECT name FROM emp1 EXCEPT SELECT name FROM emp2");
@@ -98,10 +101,14 @@ public class SetOpIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testExceptFromEmpty() throws Exception {
- populateTables();
+ createTables();
+ createCacheCopyAsReplicated("emp1");
+ createCacheCopyAsReplicated("emp2");
- copyCacheAsReplicated("emp1");
- copyCacheAsReplicated("emp2");
+ populateEmp1(client.cache("emp1"));
+ populateEmp2(client.cache("emp2"));
+ populateEmp1(client.cache("emp1Replicated"));
+ populateEmp2(client.cache("emp2Replicated"));
List<List<?>> rows = executeSql("SELECT name FROM emp1 WHERE salary < 0 EXCEPT SELECT name FROM emp2");
@@ -115,7 +122,9 @@ public class SetOpIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testExceptSeveralColumns() throws Exception {
- populateTables();
+ createTables();
+ populateEmp1(client.cache("emp1"));
+ populateEmp2(client.cache("emp2"));
List<List<?>> rows = executeSql("SELECT name, salary FROM emp1 EXCEPT SELECT name, salary FROM emp2");
@@ -127,7 +136,9 @@ public class SetOpIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testExceptAll() throws Exception {
- populateTables();
+ createTables();
+ populateEmp1(client.cache("emp1"));
+ populateEmp2(client.cache("emp2"));
List<List<?>> rows = executeSql("SELECT name FROM emp1 EXCEPT ALL SELECT name FROM emp2");
@@ -139,7 +150,9 @@ public class SetOpIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testExceptNested() throws Exception {
- populateTables();
+ createTables();
+ populateEmp1(client.cache("emp1"));
+ populateEmp2(client.cache("emp2"));
List<List<?>> rows =
executeSql("SELECT name FROM emp1 EXCEPT (SELECT name FROM emp1 EXCEPT SELECT name FROM emp2)");
@@ -152,8 +165,12 @@ public class SetOpIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testExceptReplicatedWithPartitioned() throws Exception {
- populateTables();
- copyCacheAsReplicated("emp1");
+ createTables();
+ createCacheCopyAsReplicated("emp1");
+
+ populateEmp1(client.cache("emp1"));
+ populateEmp2(client.cache("emp2"));
+ populateEmp1(client.cache("emp1Replicated"));
List<List<?>> rows = executeSql("SELECT name FROM emp1_repl EXCEPT ALL SELECT name FROM emp2");
@@ -165,9 +182,14 @@ public class SetOpIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testExceptReplicated() throws Exception {
- populateTables();
- copyCacheAsReplicated("emp1");
- copyCacheAsReplicated("emp2");
+ createTables();
+ createCacheCopyAsReplicated("emp1");
+ createCacheCopyAsReplicated("emp2");
+
+ populateEmp1(client.cache("emp1"));
+ populateEmp2(client.cache("emp2"));
+ populateEmp1(client.cache("emp1Replicated"));
+ populateEmp2(client.cache("emp2Replicated"));
List<List<?>> rows = executeSql("SELECT name FROM emp1_repl EXCEPT ALL SELECT name FROM emp2_repl");
@@ -179,8 +201,12 @@ public class SetOpIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testExceptMerge() throws Exception {
- populateTables();
- copyCacheAsReplicated("emp1");
+ createTables();
+ createCacheCopyAsReplicated("emp1");
+
+ populateEmp1(client.cache("emp1"));
+ populateEmp2(client.cache("emp2"));
+ populateEmp1(client.cache("emp1Replicated"));
List<List<?>> rows = executeSql("SELECT name FROM emp1_repl EXCEPT ALL SELECT name FROM emp2 EXCEPT ALL " +
"SELECT name FROM emp1 WHERE salary < 11");
@@ -193,20 +219,20 @@ public class SetOpIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testSetOpBigBatch() throws Exception {
- client.getOrCreateCache(new CacheConfiguration<Integer, Integer>()
+ client.getOrCreateCache(this.<Integer, Integer>cacheConfiguration()
.setName("cache1")
.setQueryEntities(F.asList(new QueryEntity(Integer.class, Integer.class).setTableName("table1")))
.setBackups(2)
);
- client.getOrCreateCache(new CacheConfiguration<Integer, Integer>()
+ client.getOrCreateCache(this.<Integer, Integer>cacheConfiguration()
.setName("cache2")
.setQueryEntities(F.asList(new QueryEntity(Integer.class, Integer.class).setTableName("table2")))
.setBackups(1)
);
- copyCacheAsReplicated("cache1");
- copyCacheAsReplicated("cache2");
+ createCacheCopyAsReplicated("cache1");
+ createCacheCopyAsReplicated("cache2");
try (IgniteDataStreamer<Integer, Integer> ds1 = client.dataStreamer("cache1");
IgniteDataStreamer<Integer, Integer> ds2 = client.dataStreamer("cache2");
@@ -330,7 +356,9 @@ public class SetOpIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testIntersect() throws Exception {
- populateTables();
+ createTables();
+ populateEmp1(client.cache("emp1"));
+ populateEmp2(client.cache("emp2"));
List<List<?>> rows = executeSql("SELECT name FROM emp1 INTERSECT SELECT name FROM emp2");
@@ -342,7 +370,9 @@ public class SetOpIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testInstersectAll() throws Exception {
- populateTables();
+ createTables();
+ populateEmp1(client.cache("emp1"));
+ populateEmp2(client.cache("emp2"));
List<List<?>> rows = executeSql("SELECT name FROM emp1 INTERSECT ALL SELECT name FROM emp2");
@@ -354,10 +384,15 @@ public class SetOpIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testIntersectEmpty() throws Exception {
- populateTables();
+ createTables();
+ createCacheCopyAsReplicated("emp1");
+ createCacheCopyAsReplicated("emp2");
+
+ populateEmp1(client.cache("emp1"));
+ populateEmp2(client.cache("emp2"));
- copyCacheAsReplicated("emp1");
- copyCacheAsReplicated("emp2");
+ populateEmp1(client.cache("emp1Replicated"));
+ populateEmp2(client.cache("emp2Replicated"));
List<List<?>> rows = executeSql("SELECT name FROM emp1 WHERE salary < 0 INTERSECT SELECT name FROM emp2");
@@ -371,8 +406,12 @@ public class SetOpIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testIntersectMerge() throws Exception {
- populateTables();
- copyCacheAsReplicated("emp1");
+ createTables();
+ createCacheCopyAsReplicated("emp1");
+
+ populateEmp1(client.cache("emp1"));
+ populateEmp2(client.cache("emp2"));
+ populateEmp1(client.cache("emp1Replicated"));
List<List<?>> rows = executeSql("SELECT name FROM emp1_repl INTERSECT ALL SELECT name FROM emp2 INTERSECT ALL " +
"SELECT name FROM emp1 WHERE salary < 14");
@@ -384,9 +423,14 @@ public class SetOpIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testIntersectReplicated() throws Exception {
- populateTables();
- copyCacheAsReplicated("emp1");
- copyCacheAsReplicated("emp2");
+ createTables();
+ createCacheCopyAsReplicated("emp1");
+ createCacheCopyAsReplicated("emp2");
+
+ populateEmp1(client.cache("emp1"));
+ populateEmp2(client.cache("emp2"));
+ populateEmp1(client.cache("emp1Replicated"));
+ populateEmp2(client.cache("emp2Replicated"));
List<List<?>> rows = executeSql("SELECT name FROM emp1_repl INTERSECT ALL SELECT name FROM emp2_repl");
@@ -398,8 +442,12 @@ public class SetOpIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testIntersectReplicatedWithPartitioned() throws Exception {
- populateTables();
- copyCacheAsReplicated("emp1");
+ createTables();
+ createCacheCopyAsReplicated("emp1");
+
+ populateEmp1(client.cache("emp1"));
+ populateEmp2(client.cache("emp2"));
+ populateEmp1(client.cache("emp1Replicated"));
List<List<?>> rows = executeSql("SELECT name FROM emp1_repl INTERSECT ALL SELECT name FROM emp2");
@@ -411,7 +459,9 @@ public class SetOpIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testIntersectSeveralColumns() throws Exception {
- populateTables();
+ createTables();
+ populateEmp1(client.cache("emp1"));
+ populateEmp2(client.cache("emp2"));
List<List<?>> rows = executeSql("SELECT name, salary FROM emp1 INTERSECT ALL SELECT name, salary FROM emp2");
@@ -423,8 +473,8 @@ public class SetOpIntegrationTest extends AbstractBasicIntegrationTest {
@Test
public void testSetOpColocated() {
executeSql("CREATE TABLE emp(empid INTEGER, deptid INTEGER, name VARCHAR, PRIMARY KEY(empid, deptid)) " +
- "WITH AFFINITY_KEY=deptid");
- executeSql("CREATE TABLE dept(deptid INTEGER, name VARCHAR, PRIMARY KEY(deptid))");
+ "WITH AFFINITY_KEY=deptid," + atomicity());
+ executeSql("CREATE TABLE dept(deptid INTEGER, name VARCHAR, PRIMARY KEY(deptid)) WITH " + atomicity());
executeSql("INSERT INTO emp VALUES (0, 0, 'test0'), (1, 0, 'test1'), (2, 1, 'test2')");
executeSql("INSERT INTO dept VALUES (0, 'test0'), (1, 'test1'), (2, 'test2')");
@@ -458,7 +508,7 @@ public class SetOpIntegrationTest extends AbstractBasicIntegrationTest {
*/
@Test
public void testSetOpRewindability() {
- executeSql("CREATE TABLE test(i INTEGER)");
+ executeSql("CREATE TABLE test(i INTEGER) WITH " + atomicity());
executeSql("INSERT INTO test VALUES (1), (2)");
assertQuery("SELECT (SELECT i FROM test EXCEPT SELECT test.i) FROM test")
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/SortAggregateIntegrationTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/SortAggregateIntegrationTest.java
index 569d363b29b..0fd9530784e 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/SortAggregateIntegrationTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/SortAggregateIntegrationTest.java
@@ -27,14 +27,13 @@ import org.apache.ignite.cache.QueryEntity;
import org.apache.ignite.cache.QueryIndex;
import org.apache.ignite.cache.QueryIndexType;
import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
-import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.junit.Test;
/**
* Sort aggregate integration test.
*/
-public class SortAggregateIntegrationTest extends AbstractBasicIntegrationTest {
+public class SortAggregateIntegrationTest extends AbstractBasicIntegrationTransactionalTest {
/** */
public static final int ROWS = 103;
@@ -45,6 +44,8 @@ public class SortAggregateIntegrationTest extends AbstractBasicIntegrationTest {
/** {@inheritDoc} */
@Override protected void afterTest() {
+ clearTransaction();
+
for (String cacheName : client.cacheNames())
client.cache(cacheName).clear();
}
@@ -77,7 +78,8 @@ public class SortAggregateIntegrationTest extends AbstractBasicIntegrationTest {
return super.getConfiguration(igniteInstanceName)
.setCacheConfiguration(
- new CacheConfiguration<>(part.getTableName())
+ cacheConfiguration()
+ .setName(part.getTableName())
.setAffinity(new RendezvousAffinityFunction(false, 8))
.setCacheMode(CacheMode.PARTITIONED)
.setQueryEntities(Arrays.asList(tbl1, part))
@@ -88,7 +90,7 @@ public class SortAggregateIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void mapReduceAggregate() throws InterruptedException {
- fillCacheTest(grid(0).cache("TEST"), ROWS);
+ fillCacheTest(client.cache("TEST"), ROWS);
List<List<?>> cursors = executeSql("SELECT /*+ DISABLE_RULE('HashAggregateConverterRule') */" +
"SUM(val0), SUM(val1), grp0 FROM TEST " +
@@ -110,7 +112,7 @@ public class SortAggregateIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void correctCollationsOnMapReduceSortAgg() throws InterruptedException {
- fillCacheTbl1(grid(0).cache("TEST"), ROWS);
+ fillCacheTbl1(client.cache("TEST"), ROWS);
List<List<?>> cursors = executeSql("SELECT PK FROM TBL1 WHERE col0 IN (SELECT col0 FROM TBL1)");
@@ -123,7 +125,7 @@ public class SortAggregateIntegrationTest extends AbstractBasicIntegrationTest {
*/
private void fillCacheTbl1(IgniteCache c, int rows) throws InterruptedException {
for (int i = 0; i < rows; ++i)
- c.put(i, new TestValTbl1(i));
+ put(client, c, i, new TestValTbl1(i));
awaitPartitionMapExchange();
}
@@ -134,7 +136,7 @@ public class SortAggregateIntegrationTest extends AbstractBasicIntegrationTest {
*/
private void fillCacheTest(IgniteCache c, int rows) throws InterruptedException {
for (int i = 0; i < rows; ++i)
- c.put(i, new TestValTest(i));
+ put(client, c, i, new TestValTest(i));
awaitPartitionMapExchange();
}
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/TableDmlIntegrationTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/TableDmlIntegrationTest.java
index 0e1e1435c83..e2ba926bc5c 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/TableDmlIntegrationTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/TableDmlIntegrationTest.java
@@ -34,9 +34,6 @@ import org.apache.calcite.sql.type.SqlTypeName;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.binary.BinaryObjectBuilder;
-import org.apache.ignite.cache.query.FieldsQueryCursor;
-import org.apache.ignite.cache.query.SqlFieldsQuery;
-import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.processors.query.IgniteSQLException;
import org.apache.ignite.internal.processors.query.QueryEngine;
@@ -49,14 +46,14 @@ import org.apache.ignite.testframework.GridTestUtils;
import org.junit.Test;
/** */
-public class TableDmlIntegrationTest extends AbstractBasicIntegrationTest {
+public class TableDmlIntegrationTest extends AbstractBasicIntegrationTransactionalTest {
/**
* Test verifies that already inserted by the current query data
* is not processed by this query again.
*/
@Test
public void testInsertAsSelect() {
- executeSql("CREATE TABLE test (epoch_cur int, epoch_copied int)");
+ executeSql("CREATE TABLE test (epoch_cur int, epoch_copied int) WITH " + atomicity());
executeSql("INSERT INTO test VALUES (0, 0)");
final String insertAsSelectSql = "INSERT INTO test SELECT ?, epoch_cur FROM test";
@@ -76,7 +73,8 @@ public class TableDmlIntegrationTest extends AbstractBasicIntegrationTest {
*/
@Test
public void testInsertAsSelectWithConcurrentDataModification() throws IgniteCheckedException {
- executeSql("CREATE TABLE test (id int primary key, val int) with cache_name=\"test\", value_type=\"my_type\"");
+ executeSql("CREATE TABLE test (id int primary key, val int) " +
+ "with cache_name=\"test\", value_type=\"my_type\", " + atomicity());
IgniteCache<Integer, Object> cache = grid(0).cache("test").withKeepBinary();
BinaryObjectBuilder builder = grid(0).binary().builder("my_type");
@@ -110,10 +108,9 @@ public class TableDmlIntegrationTest extends AbstractBasicIntegrationTest {
*/
@Test
public void testUpdate() {
- executeSql("CREATE TABLE test (val integer)");
+ executeSql("CREATE TABLE test (val integer) with " + atomicity());
- client.context().query().querySqlFields(
- new SqlFieldsQuery("CREATE INDEX test_val_idx ON test (val)").setSchema("PUBLIC"), false).getAll();
+ executeSql("CREATE INDEX test_val_idx ON test (val)");
for (int i = 1; i <= 4096; i++)
executeSql("INSERT INTO test VALUES (?)", i);
@@ -144,7 +141,7 @@ public class TableDmlIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testInsertPrimitiveKey() {
- grid(1).getOrCreateCache(new CacheConfiguration<Integer, CalciteQueryProcessorTest.Developer>()
+ grid(1).getOrCreateCache(this.<Integer, CalciteQueryProcessorTest.Developer>cacheConfiguration()
.setName("developer")
.setSqlSchema("PUBLIC")
.setIndexedTypes(Integer.class, CalciteQueryProcessorTest.Developer.class)
@@ -153,12 +150,7 @@ public class TableDmlIntegrationTest extends AbstractBasicIntegrationTest {
QueryEngine engine = Commons.lookupComponent(grid(1).context(), QueryEngine.class);
- List<FieldsQueryCursor<List<?>>> qry = engine.query(null, "PUBLIC",
- "INSERT INTO DEVELOPER(_key, name, projectId) VALUES (?, ?, ?)", 0, "Igor", 1);
-
- assertEquals(1, qry.size());
-
- List<List<?>> rows = qry.get(0).getAll();
+ List<List<?>> rows = executeSql("INSERT INTO DEVELOPER(_key, name, projectId) VALUES (?, ?, ?)", 0, "Igor", 1);
assertEquals(1, rows.size());
@@ -168,11 +160,11 @@ public class TableDmlIntegrationTest extends AbstractBasicIntegrationTest {
assertEqualsCollections(F.asList(1L), row);
- qry = engine.query(null, "PUBLIC", "select _key, * from DEVELOPER");
+ rows = executeSql("select _key, * from DEVELOPER");
- assertEquals(1, qry.size());
+ assertEquals(1, rows.size());
- row = F.first(qry.get(0).getAll());
+ row = F.first(rows);
assertNotNull(row);
@@ -182,7 +174,7 @@ public class TableDmlIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@Test
public void testInsertUpdateDeleteNonPrimitiveKey() throws Exception {
- client.getOrCreateCache(new CacheConfiguration<CalciteQueryProcessorTest.Key, CalciteQueryProcessorTest.Developer>()
+ client.getOrCreateCache(this.<CalciteQueryProcessorTest.Key, CalciteQueryProcessorTest.Developer>cacheConfiguration()
.setName("developer")
.setSqlSchema("PUBLIC")
.setIndexedTypes(CalciteQueryProcessorTest.Key.class, CalciteQueryProcessorTest.Developer.class)
@@ -193,61 +185,37 @@ public class TableDmlIntegrationTest extends AbstractBasicIntegrationTest {
QueryEngine engine = Commons.lookupComponent(grid(1).context(), QueryEngine.class);
- List<FieldsQueryCursor<List<?>>> qry = engine.query(null, "PUBLIC", "INSERT INTO DEVELOPER VALUES (?, ?, ?, ?)", 0, 0, "Igor", 1);
-
- assertEquals(1, qry.size());
-
- List<?> row = F.first(qry.get(0).getAll());
+ List<?> row = F.first(executeSql("INSERT INTO DEVELOPER VALUES (?, ?, ?, ?)", 0, 0, "Igor", 1));
assertNotNull(row);
assertEqualsCollections(F.asList(1L), row);
- qry = engine.query(null, "PUBLIC", "select * from DEVELOPER");
-
- assertEquals(1, qry.size());
-
- row = F.first(qry.get(0).getAll());
+ row = F.first(executeSql("select * from DEVELOPER"));
assertNotNull(row);
assertEqualsCollections(F.asList(0, 0, "Igor", 1), row);
- qry = engine.query(null, "PUBLIC", "UPDATE DEVELOPER d SET name = name || 'Roman' WHERE id = ?", 0);
-
- assertEquals(1, qry.size());
-
- row = F.first(qry.get(0).getAll());
+ row = F.first(executeSql("UPDATE DEVELOPER d SET name = name || 'Roman' WHERE id = ?", 0));
assertNotNull(row);
assertEqualsCollections(F.asList(1L), row);
- qry = engine.query(null, "PUBLIC", "select * from DEVELOPER");
-
- assertEquals(1, qry.size());
-
- row = F.first(qry.get(0).getAll());
+ row = F.first(executeSql("select * from DEVELOPER"));
assertNotNull(row);
assertEqualsCollections(F.asList(0, 0, "IgorRoman", 1), row);
- qry = engine.query(null, "PUBLIC", "DELETE FROM DEVELOPER WHERE id = ?", 0);
-
- assertEquals(1, qry.size());
-
- row = F.first(qry.get(0).getAll());
+ row = F.first(executeSql("DELETE FROM DEVELOPER WHERE id = ?", 0));
assertNotNull(row);
assertEqualsCollections(F.asList(1L), row);
- qry = engine.query(null, "PUBLIC", "select * from DEVELOPER");
-
- assertEquals(1, qry.size());
-
- row = F.first(qry.get(0).getAll());
+ row = F.first(executeSql("select * from DEVELOPER"));
assertNull(row);
}
@@ -257,7 +225,7 @@ public class TableDmlIntegrationTest extends AbstractBasicIntegrationTest {
*/
@Test
public void testInsertUpdateDeleteComplexKey() {
- executeSql("CREATE TABLE t(id INT, val VARCHAR, val2 VARCHAR, PRIMARY KEY(id, val))");
+ executeSql("CREATE TABLE t(id INT, val VARCHAR, val2 VARCHAR, PRIMARY KEY(id, val)) WITH " + atomicity());
executeSql("INSERT INTO t(id, val, val2) VALUES (1, 'a', 'b')");
assertQuery("SELECT * FROM t").returns(1, "a", "b").check();
@@ -279,11 +247,12 @@ public class TableDmlIntegrationTest extends AbstractBasicIntegrationTest {
*/
@Test
public void testMerge() {
- executeSql("CREATE TABLE test1 (a int, b varchar, c varchar)");
+ executeSql("CREATE TABLE test1 (a int, b varchar, c varchar) WITH " + atomicity());
+ executeSql("CREATE TABLE test2 (a int, b varchar) WITH " + atomicity());
+
executeSql("INSERT INTO test1 VALUES (0, 'a', '0')");
executeSql("INSERT INTO test1 VALUES (1, 'b', '1')");
- executeSql("CREATE TABLE test2 (a int, b varchar)");
executeSql("INSERT INTO test2 VALUES (0, '0')");
executeSql("INSERT INTO test2 VALUES (2, '2')");
@@ -305,11 +274,12 @@ public class TableDmlIntegrationTest extends AbstractBasicIntegrationTest {
*/
@Test
public void testMergeWhenMatched() {
- executeSql("CREATE TABLE test1 (a int, b varchar, c varchar)");
+ executeSql("CREATE TABLE test1 (a int, b varchar, c varchar) WITH " + atomicity());
+ executeSql("CREATE TABLE test2 (a int, b varchar) WITH " + atomicity());
+
executeSql("INSERT INTO test1 VALUES (0, 'a', '0')");
executeSql("INSERT INTO test1 VALUES (1, 'b', '1')");
- executeSql("CREATE TABLE test2 (a int, b varchar)");
executeSql("INSERT INTO test2 VALUES (0, '0')");
executeSql("INSERT INTO test2 VALUES (2, '2')");
@@ -329,11 +299,12 @@ public class TableDmlIntegrationTest extends AbstractBasicIntegrationTest {
*/
@Test
public void testMergeWhenNotMatched() {
- executeSql("CREATE TABLE test1 (a int, b varchar, c varchar)");
+ executeSql("CREATE TABLE test1 (a int, b varchar, c varchar) WITH " + atomicity());
+ executeSql("CREATE TABLE test2 (a int, b varchar) WITH " + atomicity());
+
executeSql("INSERT INTO test1 VALUES (0, 'a', '0')");
executeSql("INSERT INTO test1 VALUES (1, 'b', '1')");
- executeSql("CREATE TABLE test2 (a int, b varchar)");
executeSql("INSERT INTO test2 VALUES (0, '0')");
executeSql("INSERT INTO test2 VALUES (2, '2')");
@@ -354,7 +325,7 @@ public class TableDmlIntegrationTest extends AbstractBasicIntegrationTest {
*/
@Test
public void testMergeTableWithItself() {
- executeSql("CREATE TABLE test1 (a int, b int, c varchar)");
+ executeSql("CREATE TABLE test1 (a int, b int, c varchar) WITH " + atomicity());
executeSql("INSERT INTO test1 VALUES (0, 0, '0')");
String sql = "MERGE INTO test1 dst USING test1 src ON dst.a = src.a + 1 " +
@@ -379,12 +350,11 @@ public class TableDmlIntegrationTest extends AbstractBasicIntegrationTest {
*/
@Test
public void testMergeBatch() {
- executeSql("CREATE TABLE test1 (a int)");
+ executeSql("CREATE TABLE test1 (a int) WITH " + atomicity());
+ executeSql("CREATE TABLE test2 (a int, b int) WITH " + atomicity());
executeSql("INSERT INTO test1 SELECT x FROM TABLE(SYSTEM_RANGE(0, 9999))");
- executeSql("CREATE TABLE test2 (a int, b int)");
-
executeSql("INSERT INTO test2 SELECT x, 0 FROM TABLE(SYSTEM_RANGE(-5000, 4999))");
executeSql("MERGE INTO test2 dst USING test1 src ON dst.a = src.a " +
@@ -401,10 +371,10 @@ public class TableDmlIntegrationTest extends AbstractBasicIntegrationTest {
*/
@Test
public void testMergeAliases() {
- executeSql("CREATE TABLE test1 (a int, b int, c varchar)");
- executeSql("INSERT INTO test1 VALUES (0, 0, '0')");
+ executeSql("CREATE TABLE test1 (a int, b int, c varchar) WITH " + atomicity());
+ executeSql("CREATE TABLE test2 (a int, d int, e varchar) WITH " + atomicity());
- executeSql("CREATE TABLE test2 (a int, d int, e varchar)");
+ executeSql("INSERT INTO test1 VALUES (0, 0, '0')");
// Without aliases, column 'A' in insert statement is not ambiguous.
executeSql("MERGE INTO test2 USING test1 ON c = e " +
@@ -448,12 +418,12 @@ public class TableDmlIntegrationTest extends AbstractBasicIntegrationTest {
*/
@Test
public void testMergeKeysConflict() {
- executeSql("CREATE TABLE test1 (a int, b int)");
+ executeSql("CREATE TABLE test1 (a int, b int) WITH " + atomicity());
+ executeSql("CREATE TABLE test2 (a int primary key, b int) WITH " + atomicity());
+
executeSql("INSERT INTO test1 VALUES (0, 0)");
executeSql("INSERT INTO test1 VALUES (1, 1)");
- executeSql("CREATE TABLE test2 (a int primary key, b int)");
-
assertThrows("MERGE INTO test2 USING test1 ON test1.a = test2.a " +
"WHEN MATCHED THEN UPDATE SET b = test1.b + 1 " +
"WHEN NOT MATCHED THEN INSERT (a, b) VALUES (0, b)", IgniteSQLException.class,
@@ -477,7 +447,7 @@ public class TableDmlIntegrationTest extends AbstractBasicIntegrationTest {
IgniteSQLException.class,
"Object 'NON_EXISTENT_TABLE' not found");
- executeSql("CREATE TABLE PERSON(ID INT, PRIMARY KEY(id), NAME VARCHAR)");
+ executeSql("CREATE TABLE PERSON(ID INT, PRIMARY KEY(id), NAME VARCHAR) WITH " + atomicity());
assertThrows("" +
"MERGE INTO PERSON DST USING NON_EXISTENT_TABLE SRC ON DST.ID = SRC.ID" +
@@ -500,7 +470,7 @@ public class TableDmlIntegrationTest extends AbstractBasicIntegrationTest {
Stream.of(true, false).forEach(withPk -> {
try {
sql("CREATE TABLE integers(i INTEGER " + (withPk ? "PRIMARY KEY" : "") +
- " , col1 INTEGER DEFAULT 200, col2 INTEGER DEFAULT 300)");
+ " , col1 INTEGER DEFAULT 200, col2 INTEGER DEFAULT 300) WITH " + atomicity());
sql("INSERT INTO integers (i) VALUES (0)");
sql("INSERT INTO integers VALUES (1, DEFAULT, DEFAULT)");
@@ -525,6 +495,8 @@ public class TableDmlIntegrationTest extends AbstractBasicIntegrationTest {
.check();
}
finally {
+ clearTransaction();
+
sql("DROP TABLE IF EXISTS integers");
}
});
@@ -577,7 +549,7 @@ public class TableDmlIntegrationTest extends AbstractBasicIntegrationTest {
*/
@Test
public void testInsertDuplicateKey() {
- executeSql("CREATE TABLE test (a int primary key, b int)");
+ executeSql("CREATE TABLE test (a int primary key, b int) WITH " + atomicity());
executeSql("INSERT INTO test VALUES (0, 0)");
executeSql("INSERT INTO test VALUES (1, 1)");
@@ -602,8 +574,8 @@ public class TableDmlIntegrationTest extends AbstractBasicIntegrationTest {
long max = (Long)arg.get(1);
long min = (Long)arg.get(2);
- sql(String.format("CREATE TABLE TEST_SOURCE (ID INT PRIMARY KEY, VAL %s)", type));
- sql(String.format("CREATE TABLE TEST_DEST (ID INT PRIMARY KEY, VAL %s)", type));
+ sql(String.format("CREATE TABLE TEST_SOURCE (ID INT PRIMARY KEY, VAL %s) WITH " + atomicity(), type));
+ sql(String.format("CREATE TABLE TEST_DEST (ID INT PRIMARY KEY, VAL %s) WITH " + atomicity(), type));
sql("INSERT INTO TEST_SOURCE VALUES (1, 1)");
sql(String.format("INSERT INTO TEST_SOURCE VALUES (2, %d)", max));
@@ -628,6 +600,8 @@ public class TableDmlIntegrationTest extends AbstractBasicIntegrationTest {
IgniteSQLException.class, type + " overflow");
}
finally {
+ clearTransaction();
+
sql("DROP TABLE TEST_SOURCE");
sql("DROP TABLE TEST_DEST");
}
@@ -637,7 +611,7 @@ public class TableDmlIntegrationTest extends AbstractBasicIntegrationTest {
/** */
private void checkDefaultValue(String sqlType, String sqlVal, Object expectedVal) {
try {
- executeSql("CREATE TABLE test (dummy INT, val " + sqlType + " DEFAULT " + sqlVal + ")");
+ executeSql("CREATE TABLE test (dummy INT, val " + sqlType + " DEFAULT " + sqlVal + ") WITH " + atomicity());
executeSql("INSERT INTO test (dummy) VALUES (0)");
checkQueryResult("SELECT val FROM test", expectedVal);
@@ -648,6 +622,8 @@ public class TableDmlIntegrationTest extends AbstractBasicIntegrationTest {
checkQueryResult("SELECT val FROM test", expectedVal);
}
finally {
+ clearTransaction();
+
executeSql("DROP TABLE IF EXISTS test");
}
}
@@ -669,7 +645,7 @@ public class TableDmlIntegrationTest extends AbstractBasicIntegrationTest {
/** */
private void checkWrongDefault(String sqlType, String sqlVal) {
try {
- assertThrows("CREATE TABLE test (val " + sqlType + " DEFAULT " + sqlVal + ")",
+ assertThrows("CREATE TABLE test (val " + sqlType + " DEFAULT " + sqlVal + ") WITH " + atomicity(),
IgniteSQLException.class, "Cannot convert literal");
}
finally {
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/UserDefinedFunctionsIntegrationTransactionalTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/UserDefinedFunctionsIntegrationTransactionalTest.java
new file mode 100644
index 00000000000..c1d9d1b5438
--- /dev/null
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/UserDefinedFunctionsIntegrationTransactionalTest.java
@@ -0,0 +1,46 @@
+/*
+ * 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.calcite.integration;
+
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.QueryEntity;
+import org.apache.ignite.internal.util.typedef.F;
+import org.junit.Test;
+
+/**
+ * Integration test for user defined functions.
+ */
+public class UserDefinedFunctionsIntegrationTransactionalTest extends AbstractBasicIntegrationTransactionalTest {
+ /** */
+ @Test
+ public void testFunctions() {
+ // Cache with PUBLIC schema.
+ IgniteCache<Integer, Employer> emp3 = client.getOrCreateCache(this.<Integer, Employer>cacheConfiguration()
+ .setName("emp3")
+ .setSqlFunctionClasses(UserDefinedFunctionsIntegrationTest.OtherFunctionsLibrary.class)
+ .setSqlSchema("PUBLIC")
+ .setQueryEntities(F.asList(new QueryEntity(Integer.class, Employer.class).setTableName("emp3")))
+ );
+
+ put(client, emp3, 1, new Employer("Igor3", 100d));
+ put(client, emp3, 2, new Employer("Roman3", 200d));
+
+ assertQuery("SELECT sq(salary) FROM emp3").returns(10_000d).returns(40_000d).check();
+ assertQuery("SELECT echo(name) FROM emp3").returns("Igor3").returns("Roman3").check();
+ }
+}
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/PlannerTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/PlannerTest.java
index baf2809de54..ab1c61a2cd3 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/PlannerTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/PlannerTest.java
@@ -425,7 +425,8 @@ public class PlannerTest extends AbstractPlannerTest {
NoOpMemoryTracker.INSTANCE,
NoOpIoTracker.INSTANCE,
0,
- Commons.parametersMap(ctx.parameters()));
+ Commons.parametersMap(ctx.parameters()),
+ null);
return new LogicalRelImplementor<>(ectx, c -> r -> 0, mailboxRegistry, exchangeSvc,
new TestFailureProcessor(kernal)).go(fragment.root());
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/tx/SqlTransactionsIsolationTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/tx/SqlTransactionsIsolationTest.java
new file mode 100644
index 00000000000..de7af2a542f
--- /dev/null
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/tx/SqlTransactionsIsolationTest.java
@@ -0,0 +1,1196 @@
+/*
+ * 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.tx;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.function.BiConsumer;
+import java.util.function.IntConsumer;
+import java.util.function.IntFunction;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import java.util.stream.LongStream;
+import javax.cache.processor.EntryProcessor;
+import javax.cache.processor.EntryProcessorException;
+import javax.cache.processor.MutableEntry;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.CacheWriteSynchronizationMode;
+import org.apache.ignite.cache.QueryEntity;
+import org.apache.ignite.cache.QueryIndex;
+import org.apache.ignite.cache.query.SqlFieldsQuery;
+import org.apache.ignite.calcite.CalciteQueryEngineConfiguration;
+import org.apache.ignite.client.ClientCache;
+import org.apache.ignite.client.ClientTransaction;
+import org.apache.ignite.client.Config;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.ClientConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.client.thin.TcpIgniteClient;
+import org.apache.ignite.internal.processors.query.QueryUtils;
+import org.apache.ignite.internal.util.lang.RunnableX;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.lang.IgniteBiTuple;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.Transaction;
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionIsolation;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static java.lang.String.format;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.ignite.internal.processors.tx.SqlTransactionsIsolationTest.ModifyApi.CACHE;
+import static org.apache.ignite.internal.processors.tx.SqlTransactionsIsolationTest.ModifyApi.ENTRY_PROCESSOR;
+import static org.apache.ignite.internal.processors.tx.SqlTransactionsIsolationTest.ModifyApi.SQL;
+import static org.apache.ignite.testframework.GridTestUtils.assertThrows;
+import static org.apache.ignite.testframework.GridTestUtils.runAsync;
+import static org.apache.ignite.transactions.TransactionIsolation.READ_COMMITTED;
+
+/** */
+@RunWith(Parameterized.class)
+public class SqlTransactionsIsolationTest extends GridCommonAbstractTest {
+ /** */
+ public static final String USERS = "USERS";
+
+ /** */
+ public static final String DEPARTMENTS = "DEPARTMENTS";
+
+ /** */
+ public static final String TBL = "TBL";
+
+ /** */
+ public enum ExecutorType {
+ /** */
+ SERVER,
+
+ /** */
+ CLIENT,
+
+ /** */
+ THIN
+ }
+
+ /** */
+ public enum ModifyApi {
+ /** */
+ CACHE,
+
+ /** */
+ ENTRY_PROCESSOR,
+
+ /** */
+ SQL
+ }
+
+ /** */
+ public static final User JOHN = new User(1, 0, "John Connor");
+
+ /** */
+ public static final User SARAH = new User(2, 0, "Sarah Connor");
+
+ /** */
+ public static final User KYLE = new User(3, 0, "Kyle Reese");
+
+ /** */
+ public static final int TX_TIMEOUT = 60_000;
+
+ /** */
+ public static final int TX_SIZE = 10;
+
+ /** */
+ @Parameterized.Parameter()
+ public ModifyApi modify;
+
+ /** */
+ @Parameterized.Parameter(1)
+ public ExecutorType type;
+
+ /** */
+ @Parameterized.Parameter(2)
+ public boolean partitionAwareness;
+
+ /** */
+ @Parameterized.Parameter(3)
+ public CacheMode mode;
+
+ /** */
+ @Parameterized.Parameter(4)
+ public int gridCnt;
+
+ /** */
+ @Parameterized.Parameter(5)
+ public int backups;
+
+ /** */
+ @Parameterized.Parameter(6)
+ public boolean commit;
+
+ /** */
+ @Parameterized.Parameter(7)
+ public boolean multi;
+
+ /** */
+ @Parameterized.Parameter(8)
+ public TransactionConcurrency txConcurrency;
+
+ /** */
+ private static IgniteEx srv;
+
+ /** */
+ private static IgniteEx cli;
+
+ /** */
+ private static IgniteClient thinCli;
+
+ /** */
+ private final TransactionIsolation txIsolation = READ_COMMITTED;
+
+ /** */
+ private final Map<Integer, Set<Integer>> partsToKeys = new HashMap<>();
+
+ /** @return Test parameters. */
+ @Parameterized.Parameters(
+ name = "modify={0},qryExecutor={1},partitionAwareness={2},mode={3},gridCnt={4},backups={5},commit={6},multi={7},txConcurrency={8}")
+ public static Collection<?> parameters() {
+ List<Object[]> params = new ArrayList<>();
+
+ for (ModifyApi modify : ModifyApi.values()) {
+ for (CacheMode cacheMode : CacheMode.values()) {
+ for (int gridCnt : new int[]{1, 3, 5}) {
+ int[] backups = gridCnt > 1
+ ? new int[]{1, gridCnt - 1}
+ : new int[]{0};
+
+ for (int backup: backups) {
+ for (boolean commit : new boolean[]{false, true}) {
+ for (boolean mutli : new boolean[] {false, true}) {
+ for (TransactionConcurrency txConcurrency : TransactionConcurrency.values()) {
+ for (ExecutorType execType : new ExecutorType[]{ExecutorType.SERVER, ExecutorType.CLIENT}) {
+ params.add(new Object[]{
+ modify,
+ execType,
+ false, //partitionAwareness
+ cacheMode,
+ gridCnt,
+ backup,
+ commit,
+ mutli,
+ txConcurrency
+ });
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
+ return params;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.getSqlConfiguration().setQueryEnginesConfiguration(new CalciteQueryEngineConfiguration());
+ cfg.getTransactionConfiguration().setTxAwareQueriesEnabled(true);
+
+ return cfg;
+ }
+
+ /** */
+ private void init() throws Exception {
+ srv = startGrids(gridCnt);
+ cli = startClientGrid("client");
+ thinCli = TcpIgniteClient.start(new ClientConfiguration()
+ .setAddresses(Config.SERVER)
+ .setPartitionAwarenessEnabled(partitionAwareness));
+
+ cli.createCache(new CacheConfiguration<>(DEFAULT_CACHE_NAME).setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL));
+
+ for (CacheMode mode : CacheMode.values()) {
+ String users = tableName(USERS, mode);
+ String deps = tableName(DEPARTMENTS, mode);
+ String tbl = tableName(TBL, mode);
+
+ LinkedHashMap<String, String> userFlds = new LinkedHashMap<>();
+
+ userFlds.put("id", Integer.class.getName());
+ userFlds.put("userId", Integer.class.getName());
+ userFlds.put("departmentId", Integer.class.getName());
+ userFlds.put("fio", String.class.getName());
+
+ LinkedHashMap<String, String> depFlds = new LinkedHashMap<>();
+
+ depFlds.put("id", Integer.class.getName());
+ depFlds.put("name", String.class.getName());
+
+ cli.createCache(new CacheConfiguration<>()
+ .setName(users)
+ .setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL)
+ .setCacheMode(mode)
+ .setBackups(backups)
+ .setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC)
+ .setSqlSchema(QueryUtils.DFLT_SCHEMA)
+ .setQueryEntities(Collections.singleton(new QueryEntity()
+ .setTableName(users)
+ .setKeyType(Integer.class.getName())
+ .setValueType(User.class.getName())
+ .setKeyFieldName("id")
+ .setFields(userFlds)
+ .setIndexes(Arrays.asList(
+ new QueryIndex()
+ .setName("IDX_FIO_" + users)
+ .setFieldNames(Collections.singleton("fio"), true).setInlineSize(Character.BYTES * 20),
+ new QueryIndex()
+ .setName("IDX_USERID_" + users)
+ .setFieldNames(Collections.singleton("userId"), true)
+ )))));
+
+ cli.createCache(new CacheConfiguration<>()
+ .setName(deps)
+ .setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL)
+ .setCacheMode(mode)
+ .setBackups(backups)
+ .setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC)
+ .setSqlSchema(QueryUtils.DFLT_SCHEMA)
+ .setQueryEntities(Collections.singleton(new QueryEntity()
+ .setTableName(deps)
+ .setKeyType(Integer.class.getName())
+ .setValueType(String.class.getName())
+ .setKeyFieldName("id")
+ .setFields(depFlds))));
+
+ cli.createCache(new CacheConfiguration<Integer, Integer>()
+ .setName(tbl)
+ .setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL)
+ .setBackups(backups)
+ .setCacheMode(this.mode)
+ .setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC)
+ .setSqlSchema(QueryUtils.DFLT_SCHEMA)
+ .setQueryEntities(Collections.singleton(new QueryEntity()
+ .setTableName(tbl)
+ .setKeyType(Long.class.getName())
+ .setValueType(Long.class.getName()))));
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ super.afterTestsStopped();
+
+ stopAllGrids();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ if (F.isEmpty(Ignition.allGrids()))
+ init();
+
+ cli.cache(users()).removeAll();
+ cli.cache(tbl()).removeAll();
+ cli.cache(departments()).removeAll();
+
+ insert(F.t(1, JOHN));
+ }
+
+ /** */
+ @Test
+ public void testIndexScan() {
+ delete(1);
+
+ assertUsersSize(0);
+
+ int stepCnt = 7;
+ int stepSz = 12;
+ int outOfTxSz = 3;
+ int inTxSz = 9;
+
+ assert outOfTxSz + inTxSz == stepSz;
+
+ for (int i = 0; i < stepCnt; i++) {
+ int outOfTxStart = i * stepSz;
+
+ for (int j = 0; j < outOfTxSz; j++) {
+ int id = outOfTxStart + j;
+
+ insert(F.t(id, new User(id, 0, "User" + j))); // Intentionally repeat FIO to make same indexed keys.
+ }
+ }
+
+ assertUsersSize(stepCnt * outOfTxSz);
+
+ checkQueryWithPartitionFilter();
+
+ insideTx(() -> {
+ int nullFio = 0;
+
+ for (int i = 0; i < stepCnt; i++) {
+ int outOfTxStart = i * stepSz;
+ int inTxStart = i * stepSz + outOfTxSz;
+
+ assertEquals(
+ outOfTxStart,
+ sql(format("SELECT MIN(userid) FROM %s WHERE userid >= ?", users()), outOfTxStart).get(0).get(0)
+ );
+ assertEquals(
+ outOfTxStart + 1,
+ sql(format("SELECT MIN(userid) FROM %s WHERE userid > ?", users()), outOfTxStart).get(0).get(0)
+ );
+
+ for (int j = 0; j < inTxSz; j++) {
+ final int id = inTxStart + j;
+
+ String fio = ThreadLocalRandom.current().nextBoolean()
+ ? ("User" + j) // Intentionally repeat FIO to make same indexed keys.
+ : null;
+
+ if (fio == null)
+ nullFio++;
+
+ insert(F.t(id, new User(id, 0, fio)));
+
+ if (txConcurrency == TransactionConcurrency.OPTIMISTIC) {
+ // Concurrent query must not see any transaction data.
+ runAsync(() -> {
+ RunnableX check = () -> {
+ assertUsersSize(stepCnt * outOfTxSz);
+
+ assertNull(select(id, CACHE));
+ assertNull(select(id, SQL));
+ };
+
+ insideTx(check, false);
+ check.run();
+ }).get(TX_TIMEOUT);
+ }
+
+ long expTblSz = (long)(stepCnt * outOfTxSz) + i * inTxSz + j + 1;
+
+ assertUsersSize(expTblSz);
+
+ List<List<?>> rows = sql(format("SELECT fio FROM %s ORDER BY fio", users()));
+
+ assertEquals(expTblSz, rows.size());
+
+ ensureSorted(rows, true);
+
+ rows = sql(format("SELECT COUNT(fio) FROM %s", users()));
+
+ assertEquals(expTblSz - nullFio, rows.get(0).get(0));
+
+ assertEquals(
+ id,
+ sql(format("SELECT MIN(userid) FROM %s WHERE userid BETWEEN ? AND ?", users()), id, 500).get(0).get(0)
+ );
+ }
+
+ checkQueryWithPartitionFilter();
+
+ List<List<?>> res = sql(
+ format("SELECT id FROM %s WHERE userId >= ? AND userId <= ? ORDER BY id LIMIT ?", users()),
+ i * stepSz,
+ (i + 1) * stepSz,
+ stepSz
+ );
+
+ assertEquals(stepSz, res.size());
+
+ for (int j = 0; j < stepSz; j++)
+ assertEquals(outOfTxStart + j, res.get(j).get(0));
+
+ int offset = 3;
+
+ res = sql(
+ format("SELECT id FROM %s WHERE userId >= ? AND userId <= ? ORDER BY id LIMIT ? OFFSET ?", users()),
+ i * stepSz,
+ (i + 1) * stepSz,
+ stepSz - offset,
+ offset
+ );
+
+ assertEquals(stepSz - offset, res.size());
+
+ for (int j = 0; j < stepSz - offset; j++)
+ assertEquals(outOfTxStart + j + offset, res.get(j).get(0));
+ }
+
+ for (int i = 0; i < stepCnt; i++) {
+ int start = i * stepSz;
+
+ for (int j = 0; j < outOfTxSz; j++) {
+ int id = start + j;
+
+ delete(id);
+
+ long expTblSz = (stepCnt * stepSz) - (i * outOfTxSz + j + 1);
+
+ assertUsersSize(expTblSz);
+
+ List<List<?>> rows = sql(format("SELECT fio FROM %s ORDER BY fio DESC", users()));
+
+ assertEquals(expTblSz, rows.size());
+
+ ensureSorted(rows, false);
+ }
+ }
+
+ checkQueryWithPartitionFilter();
+ }, true);
+
+ assertUsersSize(inTxSz * stepCnt);
+ }
+
+ /** */
+ @Test
+ public void testJoin() {
+ partsToKeys.clear();
+
+ sql(format("DELETE FROM %s", users()));
+ sql(format("DELETE FROM %s", departments()));
+
+ assertUsersSize(0);
+ assertTableSize(0, departments());
+
+ int depCnt = 5;
+ int userCnt = 5;
+
+ Map<Integer, List<User>> data = new HashMap<>();
+
+ IntFunction<String> uname = i -> "User " + i + 1;
+
+ IntConsumer insertDepartment = i -> {
+ data.put(i + 1, new ArrayList<>());
+
+ sql(format("INSERT INTO %s(ID, NAME) VALUES(?, ?)", departments()), i + 1, (i + 1) + " department");
+ };
+
+ BiConsumer<Integer, Integer> insertUser = (id, depId) -> {
+ User user = new User(id, depId, uname.apply(depId));
+
+ insert(F.t(id, user));
+
+ data.get(depId).add(user);
+ };
+
+ for (int i = 0; i < depCnt; i++)
+ insertDepartment.accept(i);
+
+ for (int i = 0; i < depCnt; i++) {
+ for (int j = 0; j < userCnt; j++)
+ insertUser.accept(i * depCnt + j, j + 1);
+ }
+
+ insertDepartment.accept(depCnt);
+
+ checkJoin(data, uname);
+
+ insideTx(() -> {
+ for (int i = depCnt + 1; i < depCnt * 2; i++)
+ insertDepartment.accept(i);
+
+ for (int i = 0; i < depCnt * 2; i++) {
+ for (int j = 0; j < userCnt; j++) {
+ if (i >= depCnt)
+ insertUser.accept(i * depCnt + j, i + 1);
+
+ checkJoin(data, uname);
+ }
+ }
+ }, false);
+ }
+
+ /** */
+ private void checkJoin(Map<Integer, List<User>> data, IntFunction<String> uname) {
+ List<List<?>> res = sql(
+ format("SELECT d.id, u.id, u.fio FROM %s d JOIN %s u ON d.id = u.departmentId", departments(), users())
+ );
+
+ Map<Integer, List<User>> data0 = new HashMap<>();
+
+ data.forEach((key, value) -> data0.put(key, new ArrayList<>(value)));
+
+ assertEquals(data0.values().stream().mapToInt(List::size).sum(), res.size());
+
+ for (List<?> row : res) {
+ List<User> users = data0.get(row.get(0));
+
+ assertNotNull(users);
+
+ assertTrue(users.removeIf(u -> {
+ if (u.userId != (Integer)row.get(1))
+ return false;
+
+ assertEquals(u.fio, row.get(2));
+
+ return true;
+ }));
+ }
+
+ data0.values().forEach(l -> assertTrue(l.isEmpty()));
+
+ res = sql(
+ format("SELECT d.id, COUNT(*) FROM %s d JOIN %s u ON d.id = u.departmentId GROUP BY d.id", departments(), users())
+ );
+
+ assertEquals((int)data.values().stream().filter(((Predicate<List<?>>)List::isEmpty).negate()).count(), res.size());
+
+ for (List<?> row : res)
+ assertEquals((long)data.get(row.get(0)).size(), row.get(1));
+
+ res = sql(
+ format("SELECT d.id, COUNT(*) FROM %s d LEFT JOIN %s u ON d.id = u.departmentId GROUP BY d.id", departments(), users())
+ );
+
+ assertEquals(data.size(), res.size());
+
+ for (List<?> row : res) {
+ long size = data.get(row.get(0)).size();
+
+ assertEquals(size == 0 ? 1 : size, row.get(1));
+ }
+
+ res = sql(
+ format("SELECT d.id, u.fio FROM %s d JOIN %s u ON d.id = u.departmentId", departments(), users())
+ );
+
+ assertEquals(data.values().stream().mapToInt(List::size).sum(), res.size());
+
+ for (List<?> row : res)
+ assertEquals(uname.apply((Integer)row.get(0)), row.get(1));
+ }
+
+ /** */
+ @Test
+ public void testInsert() {
+ Runnable checkBefore = () -> {
+ for (int i = 4; i <= (multi ? 6 : 4); i++) {
+ assertNull(CACHE.name(), select(i, CACHE));
+ assertNull(SQL.name(), select(i, SQL));
+ }
+ };
+
+ Runnable checkAfter = () -> {
+ for (int i = 4; i <= (multi ? 6 : 4); i++) {
+ assertEquals(CACHE.name(), JOHN, select(i, CACHE));
+ assertEquals(SQL.name(), JOHN, select(i, SQL));
+ }
+ };
+
+ checkBefore.run();
+
+ insideTx(() -> {
+ checkBefore.run();
+
+ if (multi)
+ insert(F.t(4, JOHN), F.t(5, JOHN), F.t(6, JOHN));
+ else
+ insert(F.t(4, JOHN));
+
+ if (modify == SQL) {
+ assertThrows(
+ log,
+ () -> doInsert(false, F.t(4, JOHN)),
+ IgniteException.class,
+ "Failed to INSERT some keys because they are already in cache"
+ );
+ }
+
+ checkAfter.run();
+ }, commit);
+
+ if (commit)
+ checkAfter.run();
+ else
+ checkBefore.run();
+ }
+
+ /** */
+ @Test
+ public void testUpdate() {
+ if (multi)
+ insert(F.t(2, JOHN), F.t(3, JOHN));
+
+ Runnable checkBefore = () -> {
+ for (int i = 1; i <= (multi ? 3 : 1); i++) {
+ assertEquals(JOHN, select(i, CACHE));
+ assertEquals(JOHN, select(i, SQL));
+ }
+ };
+
+ Runnable checkAfter = () -> {
+ for (int i = 1; i <= (multi ? 3 : 1); i++) {
+ assertEquals(KYLE, select(i, CACHE));
+ assertEquals(KYLE, select(i, SQL));
+ }
+ };
+
+ checkBefore.run();
+
+ insideTx(() -> {
+ checkBefore.run();
+
+ if (multi)
+ update(F.t(1, SARAH), F.t(2, SARAH), F.t(3, SARAH));
+ else
+ update(F.t(1, SARAH));
+
+ for (int i = 1; i <= (multi ? 3 : 1); i++) {
+ assertEquals(SARAH, select(i, CACHE));
+ assertEquals(SARAH, select(i, SQL));
+ }
+
+ if (multi)
+ update(F.t(1, KYLE), F.t(2, KYLE), F.t(3, KYLE));
+ else
+ update(F.t(1, KYLE));
+
+ checkAfter.run();
+ }, commit);
+
+ if (commit)
+ checkAfter.run();
+ else
+ checkBefore.run();
+ }
+
+ /** */
+ @Test
+ public void testDelete() {
+ if (multi)
+ insert(F.t(2, JOHN), F.t(3, JOHN));
+
+ Runnable checkBefore = () -> {
+ for (int i = 1; i <= (multi ? 3 : 1); i++) {
+ assertEquals(JOHN, select(i, CACHE));
+ assertEquals(JOHN, select(i, SQL));
+ }
+ };
+
+ Runnable checkAfter = () -> {
+ for (int i = 1; i <= (multi ? 3 : 1); i++) {
+ assertNull(select(i, CACHE));
+ assertNull(select(i, SQL));
+ }
+ };
+
+ checkBefore.run();
+
+ insideTx(() -> {
+ checkBefore.run();
+
+ if (multi)
+ delete(1, 2, 3);
+ else
+ delete(1);
+
+ checkAfter.run();
+ }, commit);
+
+ if (commit)
+ checkAfter.run();
+ else
+ checkBefore.run();
+ }
+
+ /** */
+ @Test
+ public void testVisibility() {
+ sql(format("DELETE FROM %s", tbl()));
+
+ assertTableSize(0, tbl());
+
+ long cnt = 100;
+
+ LongStream.range(1, 1 + cnt).forEach(i -> insideTx(() -> {
+ if (type == ExecutorType.THIN) {
+ ClientCache<Long, Long> thinCache = thinCli.cache(tbl());
+
+ thinCache.put(i, i + 1);
+
+ assertEquals("Must see transaction related data", (Long)(i + 1), thinCache.get(i));
+ }
+ else {
+ IgniteCache<Long, Long> cache = node().cache(tbl());
+
+ cache.put(i, i + 1);
+
+ assertEquals("Must see transaction related data", (Long)(i + 1), cache.get(i));
+ }
+
+ assertTableSize(i, tbl());
+ }, true));
+
+ assertTableSize(cnt, tbl());
+ }
+
+ /** */
+ private void insideTx(RunnableX test, boolean commit) {
+ if (type == ExecutorType.THIN) {
+ try (ClientTransaction tx = thinCli.transactions().txStart(txConcurrency, txIsolation, TX_TIMEOUT)) {
+ for (int i = 0; i < 3; i++)
+ thinCli.cache(DEFAULT_CACHE_NAME).put(i, i);
+
+ test.run();
+
+ if (commit) {
+ for (int i = 0; i < 3; i++)
+ thinCli.cache(DEFAULT_CACHE_NAME).remove(i, i);
+
+ tx.commit();
+ }
+ else
+ tx.rollback();
+ }
+ }
+ else {
+ Ignite initiator = node();
+
+ assertNotNull(initiator);
+
+ try (Transaction tx = initiator.transactions().txStart(txConcurrency, txIsolation, TX_TIMEOUT, TX_SIZE)) {
+ for (int i = 0; i < 3; i++)
+ initiator.cache(DEFAULT_CACHE_NAME).put(i, i);
+
+ test.run();
+
+ if (commit) {
+ for (int i = 0; i < 3; i++)
+ initiator.cache(DEFAULT_CACHE_NAME).remove(i, i);
+
+ tx.commit();
+ }
+ else
+ tx.rollback();
+ }
+ }
+ }
+
+ /** */
+ private User select(Integer id, ModifyApi api) {
+ if (api == CACHE)
+ return type == ExecutorType.THIN
+ ? (User)thinCli.cache(users()).get(id)
+ : (User)node().cache(users()).get(id);
+ else if (api == SQL) {
+ List<List<?>> res = sql(format("SELECT _VAL FROM %s WHERE _KEY = ?", users()), id);
+
+ assertNotNull(res);
+
+ return res.isEmpty() ? null : ((User)res.get(0).get(0));
+ }
+
+ fail("Unknown select: " + api);
+
+ return null;
+ }
+
+ /** */
+ private void insert(IgniteBiTuple<Integer, User>... entries) {
+ for (IgniteBiTuple<Integer, User> data : entries) {
+ assertTrue(partsToKeys
+ .computeIfAbsent(userPartition(data.get1()), k -> new HashSet<>())
+ .add(data.get1()));
+ }
+
+ doInsert(false, entries);
+ }
+
+ /** */
+ private void doInsert(boolean update, IgniteBiTuple<Integer, User>... entries) {
+ if (modify == CACHE) {
+ if (multi) {
+ Map<Integer, User> data = Arrays.stream(entries).collect(Collectors.toMap(IgniteBiTuple::get1, IgniteBiTuple::get2));
+
+ if (type == ExecutorType.THIN)
+ thinCli.cache(users()).putAll(data);
+ else
+ node().cache(users()).putAll(data);
+ }
+ else {
+ for (IgniteBiTuple<Integer, User> data : entries) {
+ if (type == ExecutorType.THIN)
+ thinCli.cache(users()).put(data.get1(), data.get2());
+ else
+ node().cache(users()).put(data.get1(), data.get2());
+ }
+ }
+ }
+ else if (modify == ENTRY_PROCESSOR) {
+ if (multi) {
+ Set<Integer> keys = Arrays.stream(entries).map(IgniteBiTuple::get1).collect(Collectors.toSet());
+ Map<Integer, User> data = Arrays.stream(entries).collect(Collectors.toMap(IgniteBiTuple::get1, IgniteBiTuple::get2));
+
+ if (type == ExecutorType.THIN)
+ thinCli.cache(users()).invokeAll(keys, new UpdateEntryProcessor<>(update), data);
+ else
+ node().cache(users()).invokeAll(keys, new UpdateEntryProcessor<>(update), data);
+ }
+ else {
+ for (IgniteBiTuple<Integer, User> data : entries) {
+ if (type == ExecutorType.THIN)
+ thinCli.cache(users()).invoke(data.get1(), new UpdateEntryProcessor<>(update), data.get2());
+ else
+ node().cache(users()).invoke(data.get1(), new UpdateEntryProcessor<>(update), data.get2());
+ }
+ }
+ }
+ else if (modify == SQL) {
+ String insert = format("INSERT INTO %s(id, userid, departmentId, fio) VALUES(?, ?, ?, ?)", users());
+
+ int colCnt = 4;
+
+ if (multi) {
+ StringBuilder sql = new StringBuilder();
+ Object[] params = new Object[entries.length * colCnt];
+
+ for (int i = 0; i < entries.length; i++) {
+ IgniteBiTuple<Integer, User> data = entries[i];
+
+ if (i != 0)
+ sql.append(";");
+
+ sql.append(insert);
+
+ params[i * colCnt] = data.get1();
+ params[i * colCnt + 1] = data.get2().userId;
+ params[i * colCnt + 2] = data.get2().departmentId;
+ params[i * colCnt + 3] = data.get2().fio;
+ }
+
+ sql(sql.toString(), params);
+ }
+ else {
+ for (IgniteBiTuple<Integer, User> data : entries)
+ sql(insert, data.get1(), data.get2().userId, data.get2().departmentId, data.get2().fio);
+ }
+ }
+ else
+ fail("Unknown insert: " + modify);
+ }
+
+ /** */
+ private void update(IgniteBiTuple<Integer, User>...entries) {
+ for (IgniteBiTuple<Integer, User> data : entries) {
+ int part = userPartition(data.get1());
+
+ assertTrue(partsToKeys.containsKey(part));
+ assertTrue(partsToKeys.get(part).contains(data.get1()));
+ }
+
+ if (modify == CACHE || modify == ENTRY_PROCESSOR)
+ doInsert(true, entries);
+ else if (modify == SQL) {
+ String update = format("UPDATE %s SET userid = ?, departmentId = ?, fio = ? WHERE id = ?", users());
+
+ int colCnt = 4;
+
+ if (multi) {
+ StringBuilder sql = new StringBuilder();
+ Object[] params = new Object[entries.length * colCnt];
+
+ for (int i = 0; i < entries.length; i++) {
+ IgniteBiTuple<Integer, User> data = entries[i];
+
+ if (i != 0)
+ sql.append(";");
+
+ sql.append(update);
+
+ params[i * colCnt] = data.get2().userId;
+ params[i * colCnt + 1] = data.get2().departmentId;
+ params[i * colCnt + 2] = data.get2().fio;
+ params[i * colCnt + 3] = data.get1();
+ }
+
+ sql(sql.toString(), params);
+ }
+ else {
+ for (IgniteBiTuple<Integer, User> data : entries)
+ sql(update, data.get2().userId, data.get2().departmentId, data.get2().fio, data.get1());
+ }
+ }
+ else
+ fail("Unknown update: " + modify);
+ }
+
+ /** */
+ private void delete(int... keys) {
+ for (int key : keys) {
+ int part = userPartition(key);
+
+ assertTrue(partsToKeys.containsKey(part));
+ assertTrue(partsToKeys.get(part).remove(key));
+ }
+
+ if (modify == CACHE) {
+ if (multi) {
+ Set<Integer> toRemove = Arrays.stream(keys).boxed().collect(Collectors.toSet());
+
+ if (type == ExecutorType.THIN)
+ thinCli.cache(users()).removeAll(toRemove);
+ else
+ node().cache(users()).removeAll(toRemove);
+ }
+ else {
+ for (int id : keys) {
+ if (type == ExecutorType.THIN)
+ thinCli.cache(users()).remove(id);
+ else
+ node().cache(users()).remove(id);
+ }
+ }
+ }
+ else if (modify == ENTRY_PROCESSOR) {
+ if (multi) {
+ Set<Integer> toRemove = Arrays.stream(keys).boxed().collect(Collectors.toSet());
+
+ if (type == ExecutorType.THIN)
+ thinCli.cache(users()).invokeAll(toRemove, new RemoveEntryProcessor<>());
+ else
+ node().cache(users()).invokeAll(toRemove, new RemoveEntryProcessor<>());
+ }
+ else {
+ for (int id : keys) {
+ if (type == ExecutorType.THIN)
+ thinCli.cache(users()).invoke(id, new RemoveEntryProcessor<>());
+ else
+ node().cache(users()).invoke(id, new RemoveEntryProcessor<>());
+ }
+ }
+ }
+ else if (modify == SQL) {
+ String delete = format("DELETE FROM %s WHERE id = ?", users());
+
+ if (multi) {
+ StringBuilder sql = new StringBuilder();
+
+ for (int i = 0; i < keys.length; i++) {
+ if (i != 0)
+ sql.append(";");
+
+ sql.append(delete);
+ }
+
+ sql(sql.toString(), Arrays.stream(keys).boxed().toArray(Object[]::new));
+ }
+ else {
+ for (int id : keys)
+ sql(delete, id);
+ }
+ }
+ else
+ fail("Unknown delete: " + modify);
+ }
+
+ /** */
+ private void checkQueryWithPartitionFilter() {
+ // https://issues.apache.org/jira/browse/IGNITE-22993
+ if (mode != CacheMode.PARTITIONED)
+ return;
+
+ for (Map.Entry<Integer, Set<Integer>> partToKeys : partsToKeys.entrySet()) {
+ assertEquals(
+ (long)partToKeys.getValue().size(),
+ sql(format("SELECT COUNT(*) FROM %s", users()), new int[]{partToKeys.getKey()}).get(0).get(0)
+ );
+
+ assertEquals(
+ partToKeys.getValue().size(),
+ sql(format("SELECT * FROM %s", users()), new int[]{partToKeys.getKey()}).size()
+ );
+ }
+ }
+
+ /** */
+ private void assertUsersSize(long sz) {
+ assertTableSize(sz, users());
+ }
+
+ /** */
+ private void assertTableSize(long sz, String tbl) {
+ assertEquals(sz, sql(format("SELECT COUNT(*) FROM %s", tbl)).get(0).get(0));
+ }
+
+ /** */
+ private static void ensureSorted(List<List<?>> rows, boolean asc) {
+ for (int k = 1; k < rows.size(); k++) {
+ String fio0 = (String)rows.get(k - 1).get(0);
+ String fio1 = (String)rows.get(k).get(0);
+
+ int cmp = fio0 == null
+ ? (fio1 == null ? 0 : -1)
+ : (fio1 == null ? 1 : fio0.compareTo(fio1));
+
+ assertTrue(asc ? (cmp <= 0) : (cmp >= 0));
+ }
+ }
+
+ /** */
+ public static class User {
+ /** */
+ private final int userId;
+
+ /** */
+ private final int departmentId;
+
+ /** */
+ private final String fio;
+
+ /** */
+ public User(int userId, int departmentId, String fio) {
+ this.userId = userId;
+ this.departmentId = departmentId;
+ this.fio = fio;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ User user = (User)o;
+ return userId == user.userId && departmentId == user.departmentId && Objects.equals(fio, user.fio);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ return Objects.hash(userId, departmentId, fio);
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(User.class, this);
+ }
+ }
+
+ /** */
+ public List<List<?>> sql(String sqlText, Object... args) {
+ return sql(sqlText, null, args);
+ }
+
+ /** */
+ public List<List<?>> sql(String sqlText, int[] parts, Object... args) {
+ SqlFieldsQuery qry = new SqlFieldsQuery(sqlText)
+ .setArgs(args)
+ .setTimeout(5, SECONDS);
+
+ if (!F.isEmpty(parts))
+ qry.setPartitions(parts);
+
+ if (type == ExecutorType.THIN)
+ return thinCli.query(qry).getAll();
+
+ if (multi) {
+ return node().context().query().querySqlFields(qry, false, false).get(0).getAll();
+ }
+ else
+ return node().cache(F.first(cli.cacheNames())).query(qry).getAll();
+ }
+
+ /** */
+ private IgniteEx node() {
+ assertTrue(type == ExecutorType.CLIENT || type == ExecutorType.SERVER);
+
+ return type == ExecutorType.CLIENT ? cli : srv;
+ }
+
+ /** */
+ private int userPartition(int id) {
+ return affinity(cli.cache(users())).partition(id);
+ }
+
+ /** */
+ private String users() {
+ return tableName(USERS, mode);
+ }
+
+ /** */
+ private String departments() {
+ return tableName(DEPARTMENTS, mode);
+ }
+
+ /** */
+ private String tbl() {
+ return tableName(TBL, mode);
+ }
+
+ /** */
+ private static String tableName(String tbl, CacheMode mode) {
+ return tbl + "_" + mode;
+ }
+
+ /** */
+ public static class RemoveEntryProcessor<K, V, T> implements EntryProcessor<K, V, T> {
+ /** {@inheritDoc} */
+ @Override public T process(MutableEntry<K, V> entry, Object... arguments) throws EntryProcessorException {
+ entry.remove();
+
+ return null;
+ }
+ }
+
+ /** */
+ public static class UpdateEntryProcessor<K, V, T> implements EntryProcessor<K, V, T> {
+ /** */
+ private final boolean update;
+
+ /** */
+ public UpdateEntryProcessor() {
+ this(true);
+ }
+
+ /** */
+ public UpdateEntryProcessor(boolean update) {
+ this.update = update;
+ }
+
+ /** {@inheritDoc} */
+ @Override public T process(MutableEntry<K, V> entry, Object... arguments) throws EntryProcessorException {
+ assertEquals("Expect entry " + (update ? "" : "not") + " exists", update, entry.exists());
+
+ if (arguments[0] instanceof User)
+ entry.setValue((V)arguments[0]);
+ else
+ entry.setValue((V)((Map<Integer, User>)arguments[0]).get((Integer)entry.getKey()));
+
+ return null;
+ }
+ }
+}
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/tx/SqlTransactionsUnsupportedModesTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/tx/SqlTransactionsUnsupportedModesTest.java
new file mode 100644
index 00000000000..78fdfa9cb5f
--- /dev/null
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/tx/SqlTransactionsUnsupportedModesTest.java
@@ -0,0 +1,96 @@
+/*
+ * 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.tx;
+
+import java.util.EnumSet;
+import java.util.List;
+import javax.cache.CacheException;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.query.SqlFieldsQuery;
+import org.apache.ignite.calcite.CalciteQueryEngineConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.Transaction;
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionIsolation;
+import org.junit.Test;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.ignite.testframework.GridTestUtils.assertThrows;
+
+/** */
+public class SqlTransactionsUnsupportedModesTest extends GridCommonAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.getSqlConfiguration().setQueryEnginesConfiguration(new CalciteQueryEngineConfiguration());
+ cfg.getTransactionConfiguration().setTxAwareQueriesEnabled(igniteInstanceName.contains("0"));
+
+ return cfg;
+ }
+
+ /** */
+ @Test
+ public void testUnsupportedTransactionModes() throws Exception {
+ try (IgniteEx srv = startGrid(0)) {
+ sql("CREATE TABLE T1(ID BIGINT PRIMARY KEY, NAME VARCHAR(100))");
+
+ for (TransactionConcurrency concurrency : TransactionConcurrency.values()) {
+ EnumSet<TransactionIsolation> supported = EnumSet.of(TransactionIsolation.READ_COMMITTED);
+
+ for (TransactionIsolation isolation : TransactionIsolation.values()) {
+ try (Transaction ignored = srv.transactions().txStart(concurrency, isolation)) {
+ sql("DELETE FROM T1");
+
+ assertTrue(supported.remove(isolation));
+ }
+ catch (CacheException e) {
+ assertFalse(supported.contains(isolation));
+ }
+ }
+
+ assertTrue(supported.isEmpty());
+ }
+ }
+ }
+
+ /** */
+ @Test
+ public void testJoinFailIfConfigMismatch() throws Exception {
+ try (IgniteEx srv = startGrid(0)) {
+ for (boolean client : new boolean[] {false, true}) {
+ assertThrows(
+ null,
+ () -> client ? startClientGrid(1) : startGrid(1),
+ IgniteCheckedException.class,
+ "Transactions aware queries enabled mismatch (fix txAwareQueriesEnabled property)");
+ }
+ }
+ }
+
+ /** */
+ public List<List<?>> sql(String sqlText, Object... args) {
+ SqlFieldsQuery qry = new SqlFieldsQuery(sqlText)
+ .setArgs(args)
+ .setTimeout(5, SECONDS);
+
+ return grid(0).context().query().querySqlFields(qry, false, false).get(0).getAll();
+ }
+}
diff --git a/modules/calcite/src/test/java/org/apache/ignite/testsuites/IgniteCalciteTestSuite.java b/modules/calcite/src/test/java/org/apache/ignite/testsuites/IgniteCalciteTestSuite.java
index ed7ad797597..f296e19c92c 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/testsuites/IgniteCalciteTestSuite.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/testsuites/IgniteCalciteTestSuite.java
@@ -19,12 +19,15 @@ package org.apache.ignite.testsuites;
import org.apache.ignite.internal.processors.query.calcite.QueryCheckerTest;
import org.apache.ignite.internal.processors.query.calcite.exec.ClosableIteratorsHolderTest;
+import org.apache.ignite.internal.processors.query.calcite.exec.KeyFilteringCursorTest;
import org.apache.ignite.internal.processors.query.calcite.exec.LogicalRelImplementorTest;
import org.apache.ignite.internal.processors.query.calcite.exec.exp.IgniteSqlFunctionsTest;
import org.apache.ignite.internal.processors.query.calcite.exec.tracker.MemoryTrackerTest;
import org.apache.ignite.internal.processors.query.calcite.message.CalciteCommunicationMessageSerializationTest;
import org.apache.ignite.internal.processors.query.calcite.sql.SqlCustomParserTest;
import org.apache.ignite.internal.processors.query.calcite.sql.SqlReservedWordsTest;
+import org.apache.ignite.internal.processors.tx.SqlTransactionsIsolationTest;
+import org.apache.ignite.internal.processors.tx.SqlTransactionsUnsupportedModesTest;
import org.junit.runner.RunWith;
import org.junit.runners.Suite;
@@ -47,6 +50,10 @@ import org.junit.runners.Suite;
ScriptTestSuite.class,
CalciteCommunicationMessageSerializationTest.class,
+
+ KeyFilteringCursorTest.class,
+ SqlTransactionsIsolationTest.class,
+ SqlTransactionsUnsupportedModesTest.class
})
public class IgniteCalciteTestSuite {
}
diff --git a/modules/calcite/src/test/java/org/apache/ignite/testsuites/IntegrationTestSuite.java b/modules/calcite/src/test/java/org/apache/ignite/testsuites/IntegrationTestSuite.java
index 84dcc6fdb61..089c9bebca9 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/testsuites/IntegrationTestSuite.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/testsuites/IntegrationTestSuite.java
@@ -67,6 +67,7 @@ import org.apache.ignite.internal.processors.query.calcite.integration.TableDmlI
import org.apache.ignite.internal.processors.query.calcite.integration.TimeoutIntegrationTest;
import org.apache.ignite.internal.processors.query.calcite.integration.UserDdlIntegrationTest;
import org.apache.ignite.internal.processors.query.calcite.integration.UserDefinedFunctionsIntegrationTest;
+import org.apache.ignite.internal.processors.query.calcite.integration.UserDefinedFunctionsIntegrationTransactionalTest;
import org.apache.ignite.internal.processors.query.calcite.jdbc.JdbcCrossEngineTest;
import org.apache.ignite.internal.processors.query.calcite.jdbc.JdbcQueryTest;
import org.apache.ignite.internal.processors.query.calcite.rules.JoinCommuteRulesTest;
@@ -117,6 +118,7 @@ import org.junit.runners.Suite;
JoinIntegrationTest.class,
IntervalTest.class,
UserDefinedFunctionsIntegrationTest.class,
+ UserDefinedFunctionsIntegrationTransactionalTest.class,
CorrelatesIntegrationTest.class,
SystemViewsIntegrationTest.class,
IndexRebuildIntegrationTest.class,
diff --git a/modules/core/src/main/java/org/apache/ignite/configuration/TransactionConfiguration.java b/modules/core/src/main/java/org/apache/ignite/configuration/TransactionConfiguration.java
index 06b06159282..14de853808f 100644
--- a/modules/core/src/main/java/org/apache/ignite/configuration/TransactionConfiguration.java
+++ b/modules/core/src/main/java/org/apache/ignite/configuration/TransactionConfiguration.java
@@ -19,6 +19,8 @@ package org.apache.ignite.configuration;
import java.io.Serializable;
import java.util.ArrayList;
+import java.util.EnumSet;
+import java.util.Set;
import javax.cache.configuration.Factory;
import org.apache.ignite.internal.util.TransientSerializable;
import org.apache.ignite.internal.util.typedef.internal.S;
@@ -32,6 +34,13 @@ import org.apache.ignite.transactions.TransactionIsolation;
*/
@TransientSerializable(methodName = "transientSerializableFields")
public class TransactionConfiguration implements Serializable {
+ /**
+ * Supported levels of transaction isolation for SQL queries.
+ *
+ * @see #setTxAwareQueriesEnabled(boolean)
+ */
+ public static final Set<TransactionIsolation> TX_AWARE_QUERIES_SUPPORTED_MODES = EnumSet.of(TransactionIsolation.READ_COMMITTED);
+
/** */
private static final IgniteProductVersion TX_PME_TIMEOUT_SINCE = IgniteProductVersion.fromString("2.5.1");
@@ -100,6 +109,12 @@ public class TransactionConfiguration implements Serializable {
*/
private boolean useJtaSync;
+ /**
+ * When set to true, Ignite will execute SQL and scan queries in transaction aware mode.
+ * Default is {@code false}.
+ */
+ private boolean txAwareQueriesEnabled;
+
/**
* Empty constructor.
*/
@@ -121,6 +136,7 @@ public class TransactionConfiguration implements Serializable {
tmLookupClsName = cfg.getTxManagerLookupClassName();
txManagerFactory = cfg.getTxManagerFactory();
useJtaSync = cfg.isUseJtaSynchronization();
+ txAwareQueriesEnabled = cfg.isTxAwareQueriesEnabled();
}
/**
@@ -409,6 +425,23 @@ public class TransactionConfiguration implements Serializable {
return this;
}
+ /**
+ * @return Whether to execute SQL and scan queries in transaction aware mode.
+ */
+ public boolean isTxAwareQueriesEnabled() {
+ return txAwareQueriesEnabled;
+ }
+
+ /**
+ * @param txAwareQueriesEnabled Whether to execute SQL and scan queries in transaction aware mode.
+ * @return {@code this} for chaining.
+ */
+ public TransactionConfiguration setTxAwareQueriesEnabled(boolean txAwareQueriesEnabled) {
+ this.txAwareQueriesEnabled = txAwareQueriesEnabled;
+
+ return this;
+ }
+
/** {@inheritDoc} */
@Override public String toString() {
return S.toString(TransactionConfiguration.class, this);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
index 70e3001eb9a..a2f7251c2b3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
@@ -265,6 +265,7 @@ import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_RESTART_ENABL
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_REST_PORT_RANGE;
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_SHUTDOWN_POLICY;
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_SPI_CLASS;
+import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_TX_AWARE_QUERIES_ENABLED;
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_TX_SERIALIZABLE_ENABLED;
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_USER_NAME;
import static org.apache.ignite.internal.IgniteVersionUtils.BUILD_TSTAMP_STR;
@@ -1705,6 +1706,7 @@ public class IgniteKernal implements IgniteEx, Externalizable {
// Save transactions configuration.
add(ATTR_TX_SERIALIZABLE_ENABLED, cfg.getTransactionConfiguration().isTxSerializableEnabled());
+ add(ATTR_TX_AWARE_QUERIES_ENABLED, cfg.getTransactionConfiguration().isTxAwareQueriesEnabled());
// Supported features.
add(ATTR_IGNITE_FEATURES, IgniteFeatures.allFeatures());
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteNodeAttributes.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteNodeAttributes.java
index 16b29e1a0f0..0f938884604 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteNodeAttributes.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteNodeAttributes.java
@@ -81,6 +81,9 @@ public final class IgniteNodeAttributes {
/** Internal attribute name constant. */
public static final String ATTR_TX_SERIALIZABLE_ENABLED = ATTR_PREFIX + ".tx.serializable.enabled";
+ /** Internal attribute name constant. */
+ public static final String ATTR_TX_AWARE_QUERIES_ENABLED = ATTR_PREFIX + ".tx.aware.queries.enabled";
+
/** Internal attribute name constant. */
public static final String ATTR_JMX_PORT = ATTR_PREFIX + ".jmx.port";
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/cache/query/index/sorted/inline/InlineIndexImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/cache/query/index/sorted/inline/InlineIndexImpl.java
index 25ce7d8f34f..a31a9cf8b71 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/cache/query/index/sorted/inline/InlineIndexImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/cache/query/index/sorted/inline/InlineIndexImpl.java
@@ -19,8 +19,6 @@ package org.apache.ignite.internal.cache.query.index.sorted.inline;
import java.util.ArrayDeque;
import java.util.Collections;
-import java.util.Comparator;
-import java.util.PriorityQueue;
import java.util.Queue;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -31,12 +29,9 @@ import org.apache.ignite.IgniteException;
import org.apache.ignite.failure.FailureContext;
import org.apache.ignite.internal.cache.query.index.AbstractIndex;
import org.apache.ignite.internal.cache.query.index.SingleCursor;
-import org.apache.ignite.internal.cache.query.index.SortOrder;
import org.apache.ignite.internal.cache.query.index.sorted.DurableBackgroundCleanupIndexTreeTaskV2;
-import org.apache.ignite.internal.cache.query.index.sorted.IndexKeyDefinition;
import org.apache.ignite.internal.cache.query.index.sorted.IndexKeyTypeSettings;
import org.apache.ignite.internal.cache.query.index.sorted.IndexRow;
-import org.apache.ignite.internal.cache.query.index.sorted.IndexRowComparator;
import org.apache.ignite.internal.cache.query.index.sorted.IndexRowImpl;
import org.apache.ignite.internal.cache.query.index.sorted.IndexValueCursor;
import org.apache.ignite.internal.cache.query.index.sorted.InlineIndexRowHandler;
@@ -170,7 +165,7 @@ public class InlineIndexImpl extends AbstractIndex implements InlineIndex {
for (int i = 0; i < segmentsCnt; i++)
segmentCursors[i] = find(lower, upper, lowIncl, upIncl, i, qryCtx);
- return new SegmentedIndexCursor(segmentCursors, def);
+ return new SortedSegmentedIndexCursor(segmentCursors, def);
}
finally {
lock.readLock().unlock();
@@ -593,86 +588,8 @@ public class InlineIndexImpl extends AbstractIndex implements InlineIndex {
return def;
}
- /** Single cursor over multiple segments. The next value is chosen with the index row comparator. */
- private static class SegmentedIndexCursor implements GridCursor<IndexRow> {
- /** Cursors over segments. */
- private final Queue<GridCursor<IndexRow>> cursors;
-
- /** Comparator to compare index rows. */
- private final Comparator<GridCursor<IndexRow>> cursorComp;
-
- /** */
- private IndexRow head;
-
- /** */
- SegmentedIndexCursor(GridCursor<IndexRow>[] cursors, SortedIndexDefinition idxDef) throws IgniteCheckedException {
- cursorComp = new Comparator<GridCursor<IndexRow>>() {
- private final IndexRowComparator rowComparator = idxDef.rowComparator();
-
- private final IndexKeyDefinition[] keyDefs =
- idxDef.indexKeyDefinitions().values().toArray(new IndexKeyDefinition[0]);
-
- @Override public int compare(GridCursor<IndexRow> o1, GridCursor<IndexRow> o2) {
- try {
- int keysLen = o1.get().keysCount();
-
- for (int i = 0; i < keysLen; i++) {
- int cmp = rowComparator.compareRow(o1.get(), o2.get(), i);
-
- if (cmp != 0) {
- boolean desc = keyDefs[i].order().sortOrder() == SortOrder.DESC;
-
- return desc ? -cmp : cmp;
- }
- }
-
- return 0;
- }
- catch (IgniteCheckedException e) {
- throw new IgniteException("Failed to sort remote index rows", e);
- }
- }
- };
-
- this.cursors = cursorsQueue(cursors);
- }
-
- /** {@inheritDoc} */
- @Override public boolean next() throws IgniteCheckedException {
- if (cursors.isEmpty())
- return false;
-
- GridCursor<IndexRow> c = cursors.poll();
-
- head = c.get();
-
- if (c.next())
- cursors.add(c);
-
- return true;
- }
-
- /** {@inheritDoc} */
- @Override public IndexRow get() throws IgniteCheckedException {
- return head;
- }
-
- /** */
- protected Queue<GridCursor<IndexRow>> cursorsQueue(GridCursor<IndexRow>[] cursors)
- throws IgniteCheckedException {
- PriorityQueue<GridCursor<IndexRow>> q = new PriorityQueue<>(cursors.length, cursorComp);
-
- for (GridCursor<IndexRow> c: cursors) {
- if (c.next())
- q.add(c);
- }
-
- return q;
- }
- }
-
/** First-only, single-value-only segmented cursor. */
- private static class SingleValueSegmentedIndexCursor extends SegmentedIndexCursor {
+ private static class SingleValueSegmentedIndexCursor extends SortedSegmentedIndexCursor {
/** Ctor. */
SingleValueSegmentedIndexCursor(
GridCursor<IndexRow>[] cursors,
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/cache/query/index/sorted/inline/InlineIndexTree.java b/modules/core/src/main/java/org/apache/ignite/internal/cache/query/index/sorted/inline/InlineIndexTree.java
index 2eb0b38b867..138137c7719 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/cache/query/index/sorted/inline/InlineIndexTree.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/cache/query/index/sorted/inline/InlineIndexTree.java
@@ -29,6 +29,7 @@ import org.apache.ignite.internal.cache.query.index.sorted.IndexKeyDefinition;
import org.apache.ignite.internal.cache.query.index.sorted.IndexKeyTypeSettings;
import org.apache.ignite.internal.cache.query.index.sorted.IndexRow;
import org.apache.ignite.internal.cache.query.index.sorted.IndexRowCache;
+import org.apache.ignite.internal.cache.query.index.sorted.IndexRowComparator;
import org.apache.ignite.internal.cache.query.index.sorted.IndexRowImpl;
import org.apache.ignite.internal.cache.query.index.sorted.InlineIndexRowHandler;
import org.apache.ignite.internal.cache.query.index.sorted.InlineIndexRowHandlerFactory;
@@ -256,7 +257,7 @@ public class InlineIndexTree extends BPlusTree<IndexRow, IndexRow> {
if (inlineSize == 0) {
IndexRow currRow = getRow(io, pageAddr, idx);
- return compareFullRows(currRow, row, 0);
+ return compareFullRows(currRow, row, 0, rowHandler(), def.rowComparator());
}
int fieldOff = 0;
@@ -306,27 +307,35 @@ public class InlineIndexTree extends BPlusTree<IndexRow, IndexRow> {
if (currRow == null)
currRow = getRow(io, pageAddr, idx);
- return compareFullRows(currRow, row, keyIdx);
+ return compareFullRows(currRow, row, keyIdx, rowHandler(), def.rowComparator());
}
return 0;
}
/** */
- private int compareFullRows(IndexRow currRow, IndexRow row, int from) throws IgniteCheckedException {
+ public static int compareFullRows(
+ IndexRow currRow,
+ IndexRow row,
+ int from,
+ InlineIndexRowHandler rowHnd,
+ IndexRowComparator rowCmp
+ ) throws IgniteCheckedException {
if (currRow == row)
return 0;
- for (int i = from; i < rowHandler().indexKeyDefinitions().size(); i++) {
+ List<IndexKeyDefinition> idxKeyDefs = rowHnd.indexKeyDefinitions();
+
+ for (int i = from; i < idxKeyDefs.size(); i++) {
// If a search key is null then skip other keys (consider that null shows that we should get all
// possible keys for that comparison).
if (row.key(i) == null)
return 0;
- int c = def.rowComparator().compareRow(currRow, row, i);
+ int c = rowCmp.compareRow(currRow, row, i);
if (c != 0)
- return applySortOrder(Integer.signum(c), rowHnd.indexKeyDefinitions().get(i).order().sortOrder());
+ return applySortOrder(Integer.signum(c), idxKeyDefs.get(i).order().sortOrder());
}
return 0;
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/cache/query/index/sorted/inline/SortedSegmentedIndexCursor.java b/modules/core/src/main/java/org/apache/ignite/internal/cache/query/index/sorted/inline/SortedSegmentedIndexCursor.java
new file mode 100644
index 00000000000..5384cff40a3
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/cache/query/index/sorted/inline/SortedSegmentedIndexCursor.java
@@ -0,0 +1,108 @@
+/*
+ * 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.cache.query.index.sorted.inline;
+
+import java.util.Comparator;
+import java.util.PriorityQueue;
+import java.util.Queue;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.internal.cache.query.index.SortOrder;
+import org.apache.ignite.internal.cache.query.index.sorted.IndexKeyDefinition;
+import org.apache.ignite.internal.cache.query.index.sorted.IndexRow;
+import org.apache.ignite.internal.cache.query.index.sorted.IndexRowComparator;
+import org.apache.ignite.internal.cache.query.index.sorted.SortedIndexDefinition;
+import org.apache.ignite.internal.util.lang.GridCursor;
+
+/** Single cursor over multiple segments. The next value is chosen with the index row comparator. */
+public class SortedSegmentedIndexCursor implements GridCursor<IndexRow> {
+ /** Cursors over segments. */
+ private final Queue<GridCursor<IndexRow>> cursors;
+
+ /** Comparator to compare index rows. */
+ private final Comparator<GridCursor<IndexRow>> cursorComp;
+
+ /** */
+ private IndexRow head;
+
+ /** */
+ public SortedSegmentedIndexCursor(GridCursor<IndexRow>[] cursors, SortedIndexDefinition idxDef) throws IgniteCheckedException {
+ cursorComp = new Comparator<>() {
+ private final IndexRowComparator rowComparator = idxDef.rowComparator();
+
+ private final IndexKeyDefinition[] keyDefs =
+ idxDef.indexKeyDefinitions().values().toArray(new IndexKeyDefinition[0]);
+
+ @Override public int compare(GridCursor<IndexRow> o1, GridCursor<IndexRow> o2) {
+ try {
+ int keysLen = o1.get().keysCount();
+
+ for (int i = 0; i < keysLen; i++) {
+ int cmp = rowComparator.compareRow(o1.get(), o2.get(), i);
+
+ if (cmp != 0) {
+ boolean desc = keyDefs[i].order().sortOrder() == SortOrder.DESC;
+
+ return desc ? -cmp : cmp;
+ }
+ }
+
+ return 0;
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteException("Failed to sort remote index rows", e);
+ }
+ }
+ };
+
+ this.cursors = cursorsQueue(cursors);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean next() throws IgniteCheckedException {
+ if (cursors.isEmpty())
+ return false;
+
+ GridCursor<IndexRow> c = cursors.poll();
+
+ head = c.get();
+
+ if (c.next())
+ cursors.add(c);
+
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public IndexRow get() throws IgniteCheckedException {
+ return head;
+ }
+
+ /** */
+ protected Queue<GridCursor<IndexRow>> cursorsQueue(GridCursor<IndexRow>[] cursors)
+ throws IgniteCheckedException {
+ PriorityQueue<GridCursor<IndexRow>> q = new PriorityQueue<>(cursors.length, cursorComp);
+
+ for (GridCursor<IndexRow> c: cursors) {
+ if (c.next())
+ q.add(c);
+ }
+
+ return q;
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ValidationOnNodeJoinUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ValidationOnNodeJoinUtils.java
index b17248166fa..9f15ba184fb 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ValidationOnNodeJoinUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ValidationOnNodeJoinUtils.java
@@ -73,6 +73,7 @@ import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_ASYNC;
import static org.apache.ignite.configuration.DeploymentMode.ISOLATED;
import static org.apache.ignite.configuration.DeploymentMode.PRIVATE;
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_CONSISTENCY_CHECK_SKIPPED;
+import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_TX_AWARE_QUERIES_ENABLED;
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_TX_SERIALIZABLE_ENABLED;
import static org.apache.ignite.internal.processors.cache.GridCacheUtils.isDefaultDataRegionPersistent;
import static org.apache.ignite.internal.processors.security.SecurityUtils.nodeSecurityContext;
@@ -575,6 +576,18 @@ public class ValidationOnNodeJoinUtils {
", locTxSerializableEnabled=" + locTxCfg.isTxSerializableEnabled() +
", rmtTxSerializableEnabled=" + rmtTxSer + ']');
}
+
+ Boolean rmtTxAwareQryEnabled = rmt.attribute(ATTR_TX_AWARE_QUERIES_ENABLED);
+
+ if (rmtTxAwareQryEnabled != null) {
+ TransactionConfiguration locTxCfg = ctx.config().getTransactionConfiguration();
+
+ if (!rmtTxAwareQryEnabled.equals(locTxCfg.isTxAwareQueriesEnabled()))
+ throw new IgniteCheckedException("Transactions aware queries enabled mismatch " +
+ "(fix txAwareQueriesEnabled property) [rmtNodeId=" + rmt.id() +
+ ", locTxAwareQueriesEnabled=" + locTxCfg.isTxAwareQueriesEnabled() +
+ ", rmtTxAwareQueriesEnabled=" + rmtTxAwareQryEnabled + ']');
+ }
}
/**
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/BPlusTree.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/BPlusTree.java
index aae72871fe2..5c5f1929e6d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/BPlusTree.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/BPlusTree.java
@@ -5931,6 +5931,16 @@ public abstract class BPlusTree<L, T extends L> extends DataStructure implements
*/
public boolean apply(BPlusTree<L, T> tree, BPlusIO<L> io, long pageAddr, int idx)
throws IgniteCheckedException;
+
+ /**
+ * Optional operation to provide access to the last analyzed row.
+ * Can be used to optimize chained filters and omit several tree reads.
+ *
+ * @return Last row that was analyzed or {@code null}.
+ */
+ public default L lastRow() {
+ return null;
+ }
}
/**
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
index 5c30d4a5e04..e0f7d1c7389 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
@@ -998,7 +998,7 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
/**
* @return Transaction for current thread.
*/
- public <T> T tx() {
+ public <T extends IgniteInternalTx> T tx() {
IgniteInternalTx tx = txContext();
return tx != null ? (T)tx : (T)tx(null, Thread.currentThread().getId());
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
index 9e8897f2ef0..f5f81015c46 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
@@ -60,6 +60,7 @@ import org.apache.ignite.cache.query.SqlQuery;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.QueryEngineConfiguration;
+import org.apache.ignite.configuration.TransactionConfiguration;
import org.apache.ignite.events.CacheQueryExecutedEvent;
import org.apache.ignite.internal.GridComponent;
import org.apache.ignite.internal.GridKernalContext;
@@ -87,6 +88,7 @@ import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.QueryCursorImpl;
import org.apache.ignite.internal.processors.cache.binary.CacheObjectBinaryProcessorImpl;
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.persistence.CacheDataRow;
import org.apache.ignite.internal.processors.cache.query.CacheQueryFuture;
import org.apache.ignite.internal.processors.cache.query.GridCacheQueryType;
@@ -299,6 +301,9 @@ public class GridQueryProcessor extends GridProcessorAdapter {
/** Schema manager. */
private final SchemaManager schemaMgr;
+ /** @see TransactionConfiguration#isTxAwareQueriesEnabled() */
+ private final boolean txAwareQueriesEnabled;
+
/**
* Constructor.
*
@@ -338,6 +343,8 @@ public class GridQueryProcessor extends GridProcessorAdapter {
initQueryEngines();
idxBuildStatusStorage = new IndexBuildStatusStorage(ctx);
+
+ txAwareQueriesEnabled = ctx.config().getTransactionConfiguration().isTxAwareQueriesEnabled();
}
/** {@inheritDoc} */
@@ -3031,7 +3038,11 @@ public class GridQueryProcessor extends GridProcessorAdapter {
if (qry.isLocal() && ctx.clientNode())
throw new CacheException("Execution of local SqlFieldsQuery on client node disallowed.");
- return executeQuerySafe(cctx, () -> {
+ final GridNearTxLocal userTx = txAwareQueriesEnabled
+ ? ctx.cache().context().tm().userTx()
+ : null;
+
+ return executeQuerySafe(cctx, userTx, () -> {
final String schemaName = qry.getSchema() == null ? schemaName(cctx) : qry.getSchema();
IgniteOutClosureX<List<FieldsQueryCursor<List<?>>>> clo =
@@ -3050,9 +3061,17 @@ public class GridQueryProcessor extends GridProcessorAdapter {
failOnMultipleStmts
);
+ QueryContext qryCtx = QueryContext.of(
+ qry,
+ cliCtx,
+ cancel,
+ qryProps,
+ userTx == null ? null : userTx.xidVersion()
+ );
+
if (qry instanceof SqlFieldsQueryEx && ((SqlFieldsQueryEx)qry).isBatched()) {
res = qryEngine.queryBatched(
- QueryContext.of(qry, cliCtx, cancel, qryProps),
+ qryCtx,
schemaName,
qry.getSql(),
((SqlFieldsQueryEx)qry).batchedArguments()
@@ -3060,7 +3079,7 @@ public class GridQueryProcessor extends GridProcessorAdapter {
}
else {
res = qryEngine.query(
- QueryContext.of(qry, cliCtx, cancel, qryProps),
+ qryCtx,
schemaName,
qry.getSql(),
qry.getArgs() != null ? qry.getArgs() : X.EMPTY_OBJECT_ARRAY
@@ -3068,6 +3087,13 @@ public class GridQueryProcessor extends GridProcessorAdapter {
}
}
else {
+ if (userTx != null && txAwareQueriesEnabled) {
+ throw new IgniteSQLException(
+ "SQL aware queries are not supported by Indexing query engine",
+ IgniteQueryErrorCode.UNSUPPORTED_OPERATION
+ );
+ }
+
res = idx.querySqlFields(
schemaName,
qry,
@@ -3096,7 +3122,7 @@ public class GridQueryProcessor extends GridProcessorAdapter {
) {
checkxModuleEnabled();
- return executeQuerySafe(null, () -> {
+ return executeQuerySafe(null, null, () -> {
final String schemaName = qry.getSchema() == null ? QueryUtils.DFLT_SCHEMA : qry.getSchema();
QueryEngine qryEngine = engineForQuery(cliCtx, qry);
@@ -3124,7 +3150,7 @@ public class GridQueryProcessor extends GridProcessorAdapter {
) {
checkxModuleEnabled();
- return executeQuerySafe(null, () -> {
+ return executeQuerySafe(null, null, () -> {
final String schemaName = qry.getSchema() == null ? QueryUtils.DFLT_SCHEMA : qry.getSchema();
QueryEngine qryEngine = engineForQuery(cliCtx, qry);
@@ -3201,27 +3227,43 @@ public class GridQueryProcessor extends GridProcessorAdapter {
* Execute query setting busy lock, preserving current cache context and properly handling checked exceptions.
*
* @param cctx Cache context.
+ * @param userTx User transaction.
* @param supplier Code to be executed.
* @return Result.
*/
- private <T> T executeQuerySafe(@Nullable final GridCacheContext<?, ?> cctx, GridPlainOutClosure<T> supplier) {
+ private <T> T executeQuerySafe(
+ @Nullable final GridCacheContext<?, ?> cctx,
+ @Nullable final GridNearTxLocal userTx,
+ GridPlainOutClosure<T> supplier
+ ) {
GridCacheContext oldCctx = curCache.get();
curCache.set(cctx);
- if (!busyLock.enterBusy())
- throw new IllegalStateException("Failed to execute query (grid is stopping).");
-
try {
- return supplier.apply();
+ if (!busyLock.enterBusy())
+ throw new IllegalStateException("Failed to execute query (grid is stopping).");
+
+ if (userTx != null)
+ userTx.suspend();
+
+ try {
+ return supplier.apply();
+ }
+ catch (IgniteCheckedException e) {
+ throw new CacheException(e);
+ }
+ finally {
+ curCache.set(oldCctx);
+
+ busyLock.leaveBusy();
+
+ if (userTx != null)
+ userTx.resume();
+ }
}
catch (IgniteCheckedException e) {
- throw new CacheException(e);
- }
- finally {
- curCache.set(oldCctx);
-
- busyLock.leaveBusy();
+ throw new IgniteException(e);
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFunc.java b/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFunc.java
index 8fb3c4d8a78..8270636dc07 100755
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFunc.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFunc.java
@@ -3181,6 +3181,22 @@ public class GridFunc {
return true;
}
+ /**
+ * @param arr Array to check.
+ * @return {@code True} if array sorted, {@code false} otherwise.
+ */
+ public static boolean isSorted(int[] arr) {
+ if (isEmpty(arr) || arr.length == 1)
+ return true;
+
+ for (int i = 1; i < arr.length; i++) {
+ if (arr[i - 1] > arr[i])
+ return false;
+ }
+
+ return true;
+ }
+
/**
* Return supplier that suppress any exception throwed by {@code s}.
* Returned supplier will produce {@code 0} on any exception in {@code s}.
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/index/H2TransactionAwareQueriesEnabledTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/index/H2TransactionAwareQueriesEnabledTest.java
new file mode 100644
index 00000000000..84aee162b57
--- /dev/null
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/index/H2TransactionAwareQueriesEnabledTest.java
@@ -0,0 +1,80 @@
+/*
+ * 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.cache.index;
+
+import java.util.List;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.query.SqlFieldsQuery;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.processors.query.IgniteSQLException;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.Transaction;
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionIsolation;
+import org.junit.Test;
+
+import static org.apache.ignite.testframework.GridTestUtils.assertThrows;
+
+/** */
+public class H2TransactionAwareQueriesEnabledTest extends GridCommonAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.getTransactionConfiguration().setTxAwareQueriesEnabled(true);
+ cfg.setCacheConfiguration(new CacheConfiguration<>(DEFAULT_CACHE_NAME).setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL));
+
+ return cfg;
+ }
+
+ /** */
+ @Test
+ public void testFailOnSqlQueryInsideTransaction() throws Exception {
+ Ignite srv = startGrid(0);
+ Ignite cli = startClientGrid(1);
+
+ sql(srv, "CREATE TABLE T(ID BIGINT PRIMARY KEY, NAME VARCHAR) WITH \"atomicity=transactional\"");
+
+ awaitPartitionMapExchange();
+
+ for (Ignite node : new Ignite[]{srv, cli}) {
+ for (TransactionConcurrency txConcurrenty : TransactionConcurrency.values()) {
+ for (TransactionIsolation txIsolation : TransactionIsolation.values()) {
+ try (Transaction tx = node.transactions().txStart(txConcurrenty, txIsolation)) {
+ assertThrows(
+ null,
+ () -> sql(node, "SELECT * FROM T"),
+ IgniteSQLException.class,
+ "SQL aware queries are not supported by Indexing query engine"
+ );
+
+ tx.rollback();
+ }
+ }
+ }
+ }
+
+ }
+
+ /** */
+ public List<List<?>> sql(Ignite node, String sql) {
+ return node.cache(DEFAULT_CACHE_NAME).query(new SqlFieldsQuery(sql)).getAll();
+ }
+}
diff --git a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite3.java b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite3.java
index 286dd897116..78e2896767e 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite3.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite3.java
@@ -91,6 +91,7 @@ import org.apache.ignite.internal.processors.cache.index.H2ConnectionSettingsTes
import org.apache.ignite.internal.processors.cache.index.H2RowCachePageEvictionTest;
import org.apache.ignite.internal.processors.cache.index.H2RowCacheSelfTest;
import org.apache.ignite.internal.processors.cache.index.H2RowExpireTimeIndexSelfTest;
+import org.apache.ignite.internal.processors.cache.index.H2TransactionAwareQueriesEnabledTest;
import org.apache.ignite.internal.processors.cache.index.IgniteDecimalSelfTest;
import org.apache.ignite.internal.processors.cache.index.IndexColumnTypeMismatchTest;
import org.apache.ignite.internal.processors.cache.index.LongIndexNameTest;
@@ -375,7 +376,9 @@ import org.junit.runners.Suite;
CacheEventsCdcTest.class,
CdcIndexRebuildTest.class,
- DumpCacheConfigTest.class
+ DumpCacheConfigTest.class,
+
+ H2TransactionAwareQueriesEnabledTest.class,
})
public class IgniteBinaryCacheQueryTestSuite3 {
/** Setup lazy mode default. */