You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by vo...@apache.org on 2018/08/30 11:53:40 UTC
[18/38] ignite git commit: IGNITE-4191: MVCC and transactional SQL
support. Joint multi-man-years efforts of Semen Boikov, Igor Seliverstov,
Alexander Paschenko, Igor Sapego, Sergey Kalashnikov, Roman Kondakov,
Pavel Kuznetsov, Ivan Pavlukhin, Andrey Mas
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccDataInnerIO.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccDataInnerIO.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccDataInnerIO.java
new file mode 100644
index 0000000..0b2f91c
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccDataInnerIO.java
@@ -0,0 +1,71 @@
+/*
+ * 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.tree.mvcc.data;
+
+import org.apache.ignite.internal.pagemem.PageUtils;
+import org.apache.ignite.internal.processors.cache.persistence.CacheDataRowAdapter;
+import org.apache.ignite.internal.processors.cache.persistence.CacheSearchRow;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.IOVersions;
+import org.apache.ignite.internal.processors.cache.tree.AbstractDataInnerIO;
+import org.apache.ignite.internal.util.typedef.internal.CU;
+import org.apache.ignite.lang.IgniteInClosure;
+
+/**
+ *
+ */
+public final class MvccDataInnerIO extends AbstractDataInnerIO {
+ /** */
+ public static final IOVersions<MvccDataInnerIO> VERSIONS = new IOVersions<>(
+ new MvccDataInnerIO(1)
+ );
+
+ /**
+ * @param ver Page format version.
+ */
+ private MvccDataInnerIO(int ver) {
+ super(T_DATA_REF_MVCC_INNER, ver, true, 32);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void visit(long pageAddr, IgniteInClosure<CacheSearchRow> c) {
+ int cnt = getCount(pageAddr);
+
+ for (int i = 0; i < cnt; i++)
+ c.apply(new MvccDataRow(getLink(pageAddr, i)));
+ }
+
+ /** {@inheritDoc} */
+ @Override protected boolean storeMvccVersion() {
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getMvccCoordinatorVersion(long pageAddr, int idx) {
+ return PageUtils.getLong(pageAddr, offset(idx) + 12);
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getMvccCounter(long pageAddr, int idx) {
+ return PageUtils.getLong(pageAddr, offset(idx) + 20);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getMvccOperationCounter(long pageAddr, int idx) {
+ return PageUtils.getInt(pageAddr, offset(idx) + 28);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccDataLeafIO.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccDataLeafIO.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccDataLeafIO.java
new file mode 100644
index 0000000..ab498d9
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccDataLeafIO.java
@@ -0,0 +1,89 @@
+/*
+ * 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.tree.mvcc.data;
+
+import org.apache.ignite.internal.pagemem.PageUtils;
+import org.apache.ignite.internal.processors.cache.persistence.CacheSearchRow;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.IOVersions;
+import org.apache.ignite.internal.processors.cache.tree.AbstractDataLeafIO;
+import org.apache.ignite.lang.IgniteInClosure;
+
+/**
+ *
+ */
+public final class MvccDataLeafIO extends AbstractDataLeafIO {
+ /** */
+ public static final IOVersions<MvccDataLeafIO> VERSIONS = new IOVersions<>(
+ new MvccDataLeafIO(1)
+ );
+
+ /**
+ * @param ver Page format version.
+ */
+ private MvccDataLeafIO(int ver) {
+ super(T_DATA_REF_MVCC_LEAF, ver, 48);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void visit(long pageAddr, IgniteInClosure<CacheSearchRow> c) {
+ int cnt = getCount(pageAddr);
+
+ for (int i = 0; i < cnt; i++)
+ c.apply(new MvccDataRow(getLink(pageAddr, i)));
+ }
+
+ /** {@inheritDoc} */
+ @Override protected boolean storeMvccVersion() {
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getMvccCoordinatorVersion(long pageAddr, int idx) {
+ return PageUtils.getLong(pageAddr, offset(idx) + 12);
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getMvccCounter(long pageAddr, int idx) {
+ return PageUtils.getLong(pageAddr, offset(idx) + 20);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getMvccOperationCounter(long pageAddr, int idx) {
+ return PageUtils.getInt(pageAddr, offset(idx) + 28);
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getMvccLockCoordinatorVersion(long pageAddr, int idx) {
+ return PageUtils.getLong(pageAddr, offset(idx) + 32);
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getMvccLockCounter(long pageAddr, int idx) {
+ return PageUtils.getLong(pageAddr, offset(idx) + 40);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void setMvccLockCoordinatorVersion(long pageAddr, int idx, long lockCrd) {
+ PageUtils.putLong(pageAddr, offset(idx) + 32, lockCrd);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void setMvccLockCounter(long pageAddr, int idx, long lockCntr) {
+ PageUtils.putLong(pageAddr, offset(idx) + 40, lockCntr);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccDataRow.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccDataRow.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccDataRow.java
new file mode 100644
index 0000000..7dc2051
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccDataRow.java
@@ -0,0 +1,263 @@
+/*
+ * 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.tree.mvcc.data;
+
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.pagemem.PageUtils;
+import org.apache.ignite.internal.processors.cache.CacheGroupContext;
+import org.apache.ignite.internal.processors.cache.CacheObject;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccUtils;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccVersion;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccVersionImpl;
+import org.apache.ignite.internal.processors.cache.mvcc.txlog.TxState;
+import org.apache.ignite.internal.processors.cache.tree.DataRow;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+import org.apache.ignite.internal.util.typedef.internal.S;
+
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_COUNTER_NA;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_CRD_COUNTER_NA;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_OP_COUNTER_NA;
+import static org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPageIO.MVCC_INFO_SIZE;
+import static org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO.MVCC_HINTS_BIT_OFF;
+import static org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO.MVCC_HINTS_MASK;
+
+/**
+ *
+ */
+public class MvccDataRow extends DataRow {
+ /** Mvcc coordinator version. */
+ @GridToStringInclude
+ protected long mvccCrd;
+
+ /** Mvcc counter. */
+ @GridToStringInclude
+ protected long mvccCntr;
+
+ /** Mvcc operation counter. */
+ @GridToStringInclude
+ protected int mvccOpCntr;
+
+ /** Mvcc tx state. */
+ @GridToStringInclude
+ protected byte mvccTxState;
+
+ /** New mvcc coordinator version. */
+ @GridToStringInclude
+ protected long newMvccCrd;
+
+ /** New mvcc counter. */
+ @GridToStringInclude
+ protected long newMvccCntr;
+
+ /** New mvcc operation counter. */
+ @GridToStringInclude
+ protected int newMvccOpCntr;
+
+ /** New mvcc tx state. */
+ @GridToStringInclude
+ protected byte newMvccTxState;
+
+ /**
+ * @param link Link.
+ */
+ public MvccDataRow(long link) {
+ super(link);
+ }
+
+ /**
+ * @param grp Context.
+ * @param hash Key hash.
+ * @param link Link.
+ * @param part Partition number.
+ * @param rowData Data.
+ * @param crdVer Mvcc coordinator version.
+ * @param mvccCntr Mvcc counter.
+ * @param mvccOpCntr Mvcc operation counter.
+ */
+ public MvccDataRow(CacheGroupContext grp,
+ int hash,
+ long link,
+ int part,
+ RowData rowData,
+ long crdVer,
+ long mvccCntr,
+ int mvccOpCntr) {
+ super(grp, hash, link, part, rowData);
+
+ assert MvccUtils.mvccVersionIsValid(crdVer, mvccCntr, mvccOpCntr);
+
+ assert rowData == RowData.LINK_ONLY
+ || this.mvccCrd == crdVer && this.mvccCntr == mvccCntr && this.mvccOpCntr == mvccOpCntr :
+ "mvccVer=" + new MvccVersionImpl(crdVer, mvccCntr, mvccOpCntr) +
+ ", dataMvccVer=" + new MvccVersionImpl(this.mvccCrd, this.mvccCntr, this.mvccOpCntr) ;
+
+ if (rowData == RowData.LINK_ONLY) {
+ this.mvccCrd = crdVer;
+ this.mvccCntr = mvccCntr;
+ this.mvccOpCntr = mvccOpCntr;
+ }
+ }
+
+ /**
+ * @param key Key.
+ * @param val Value.
+ * @param ver Version.
+ * @param part Partition.
+ * @param expireTime Expire time.
+ * @param cacheId Cache ID.
+ * @param mvccVer Mvcc version.
+ * @param newMvccVer New mvcc version.
+ */
+ public MvccDataRow(KeyCacheObject key, CacheObject val, GridCacheVersion ver, int part, long expireTime, int cacheId,
+ MvccVersion mvccVer, MvccVersion newMvccVer) {
+ super(key, val, ver, part, expireTime, cacheId);
+
+ this.mvccCrd = mvccVer.coordinatorVersion();
+ this.mvccCntr = mvccVer.counter();
+ this.mvccOpCntr = mvccVer.operationCounter();
+
+ if (newMvccVer == null) {
+ newMvccCrd = MVCC_CRD_COUNTER_NA;
+ newMvccCntr = MVCC_COUNTER_NA;
+ newMvccOpCntr = MVCC_OP_COUNTER_NA;
+ }
+ else {
+ newMvccCrd = newMvccVer.coordinatorVersion();
+ newMvccCntr = newMvccVer.counter();
+ newMvccOpCntr = newMvccVer.operationCounter();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override protected int readHeader(long addr, int off) {
+ // xid_min.
+ mvccCrd = PageUtils.getLong(addr, off);
+ mvccCntr = PageUtils.getLong(addr, off + 8);
+
+ int withHint = PageUtils.getInt(addr, off + 16);
+
+ mvccOpCntr = withHint & ~MVCC_HINTS_MASK;
+ mvccTxState = (byte)(withHint >>> MVCC_HINTS_BIT_OFF);
+
+ assert MvccUtils.mvccVersionIsValid(mvccCrd, mvccCntr, mvccOpCntr);
+
+ // xid_max.
+ newMvccCrd = PageUtils.getLong(addr, off + 20);
+ newMvccCntr = PageUtils.getLong(addr, off + 28);
+
+ withHint = PageUtils.getInt(addr, off + 36);
+
+ newMvccOpCntr = withHint & ~MVCC_HINTS_MASK;
+ newMvccTxState = (byte)(withHint >>> MVCC_HINTS_BIT_OFF);
+
+ assert newMvccCrd == MVCC_CRD_COUNTER_NA || MvccUtils.mvccVersionIsValid(newMvccCrd, newMvccCntr, newMvccOpCntr);
+
+ return MVCC_INFO_SIZE;
+ }
+
+ /** {@inheritDoc} */
+ @Override public long mvccCoordinatorVersion() {
+ return mvccCrd;
+ }
+
+ /** {@inheritDoc} */
+ @Override public long mvccCounter() {
+ return mvccCntr;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int mvccOperationCounter() {
+ return mvccOpCntr;
+ }
+
+ /** {@inheritDoc} */
+ public byte mvccTxState() {
+ return mvccTxState;
+ }
+
+ /** {@inheritDoc} */
+ @Override public long newMvccCoordinatorVersion() {
+ return newMvccCrd;
+ }
+
+ /** {@inheritDoc} */
+ @Override public long newMvccCounter() {
+ return newMvccCntr;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int newMvccOperationCounter() {
+ return newMvccOpCntr;
+ }
+
+ /** {@inheritDoc} */
+ @Override public byte newMvccTxState() {
+ return newMvccTxState;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void newMvccVersion(long crd, long cntr, int opCntr) {
+ newMvccCrd = crd;
+ newMvccCntr = cntr;
+ newMvccOpCntr = opCntr;
+
+ // reset tx state
+ newMvccTxState = TxState.NA;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void mvccVersion(long crd, long cntr, int opCntr) {
+ mvccCrd = crd;
+ mvccCntr = cntr;
+ mvccOpCntr = opCntr;
+
+ // reset tx state
+ mvccTxState = TxState.NA;
+ }
+
+ /**
+ * @param mvccTxState Mvcc version Tx state hint.
+ */
+ public void mvccTxState(byte mvccTxState) {
+ this.mvccTxState = mvccTxState;
+ }
+
+ /**
+ * @param newMvccTxState New Mvcc version Tx state hint.
+ */
+ public void newMvccTxState(byte newMvccTxState) {
+ this.newMvccTxState = newMvccTxState;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int size() throws IgniteCheckedException {
+ return super.size() + MVCC_INFO_SIZE;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int headerSize() {
+ return MVCC_INFO_SIZE;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(MvccDataRow.class, this, "super", super.toString());
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccUpdateDataRow.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccUpdateDataRow.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccUpdateDataRow.java
new file mode 100644
index 0000000..81b1b58
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccUpdateDataRow.java
@@ -0,0 +1,485 @@
+/*
+ * 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.tree.mvcc.data;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.pagemem.wal.IgniteWriteAheadLogManager;
+import org.apache.ignite.internal.processors.cache.CacheObject;
+import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccUtils;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccVersion;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccVersionImpl;
+import org.apache.ignite.internal.processors.cache.mvcc.txlog.TxState;
+import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
+import org.apache.ignite.internal.processors.cache.persistence.CacheSearchRow;
+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.cache.tree.RowLinkIO;
+import org.apache.ignite.internal.processors.cache.tree.mvcc.search.MvccLinkAwareSearchRow;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.util.tostring.GridToStringExclude;
+import org.apache.ignite.internal.util.typedef.internal.S;
+
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_COUNTER_NA;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_CRD_COUNTER_NA;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_OP_COUNTER_NA;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.compare;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.isActive;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.isVisible;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.mvccVersionIsValid;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.unexpectedStateException;
+import static org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO.MVCC_HINTS_BIT_OFF;
+import static org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO.MVCC_HINTS_MASK;
+
+/**
+ *
+ */
+public class MvccUpdateDataRow extends MvccDataRow implements MvccUpdateResult, BPlusTree.TreeVisitorClosure<CacheSearchRow, CacheDataRow> {
+ /** */
+ private static final int FIRST = DIRTY << 1;
+ /** */
+ private static final int CHECK_VERSION = FIRST << 1;
+ /** */
+ private static final int LAST_COMMITTED_FOUND = CHECK_VERSION << 1;
+ /** */
+ private static final int CAN_CLEANUP = LAST_COMMITTED_FOUND << 1;
+ /** */
+ private static final int PRIMARY = CAN_CLEANUP << 1;
+ /** */
+ private static final int REMOVE_OR_LOCK = PRIMARY << 1;
+ /** */
+ private static final int NEED_HISTORY = REMOVE_OR_LOCK << 1;
+ /**
+ * During mvcc transaction processing conflicting row version could be met in storage.
+ * Not all such cases should lead to transaction abort.
+ * E.g. if UPDATE for a row meets concurrent INSERT for the same row
+ * (and row did not exist before both operations) then it means that UPDATE does not see the row at all
+ * and can proceed.
+ * This flag enables such mode when conflicting version should not lead to abort immediately
+ * but more versions should be checked.
+ */
+ private static final int FAST_UPDATE = NEED_HISTORY << 1;
+ /** */
+ private static final int FAST_MISMATCH = FAST_UPDATE << 1;
+ /** */
+ private static final int DELETED = FAST_MISMATCH << 1;
+
+ /** */
+ @GridToStringExclude
+ private final GridCacheContext cctx;
+
+ /** */
+ private ResultType res;
+
+ /** */
+ @GridToStringExclude
+ private int state;
+
+ /** */
+ private List<MvccLinkAwareSearchRow> cleanupRows;
+
+ /** */
+ private final MvccSnapshot mvccSnapshot;
+
+ /** */
+ private CacheDataRow oldRow;
+
+ /** */
+ @GridToStringExclude
+ private long resCrd;
+
+ /** */
+ @GridToStringExclude
+ private long resCntr;
+
+ /** */
+ private List<MvccLinkAwareSearchRow> historyRows;
+
+ /**
+ * @param cctx Cache context.
+ * @param key Key.
+ * @param val Value.
+ * @param ver Version.
+ * @param part Partition.
+ * @param expireTime Expire time.
+ * @param mvccSnapshot MVCC snapshot.
+ * @param newVer Update version.
+ * @param primary Primary node flag.
+ * @param lockOnly Whether no actual update should be done and the only thing to do is to acquire lock.
+ * @param needHistory Whether to collect rows created or affected by the current tx.
+ * @param fastUpdate Fast update visit mode.
+ */
+ public MvccUpdateDataRow(
+ GridCacheContext cctx,
+ KeyCacheObject key,
+ CacheObject val,
+ GridCacheVersion ver,
+ int part,
+ long expireTime,
+ MvccSnapshot mvccSnapshot,
+ MvccVersion newVer,
+ boolean primary,
+ boolean lockOnly,
+ boolean needHistory,
+ boolean fastUpdate) {
+ super(key,
+ val,
+ ver,
+ part,
+ expireTime,
+ cctx.cacheId(),
+ mvccSnapshot,
+ newVer);
+
+ this.mvccSnapshot = mvccSnapshot;
+ this.cctx = cctx;
+
+ assert !lockOnly || val == null;
+
+ int flags = FIRST;
+
+ if (primary)
+ flags |= PRIMARY | CHECK_VERSION;
+
+ if (primary && (lockOnly || val == null))
+ flags |= CAN_WRITE | REMOVE_OR_LOCK;
+
+ if (needHistory)
+ flags |= NEED_HISTORY;
+
+ if (fastUpdate)
+ flags |= FAST_UPDATE;
+
+ setFlags(flags);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int visit(BPlusTree<CacheSearchRow, CacheDataRow> tree,
+ BPlusIO<CacheSearchRow> io,
+ long pageAddr,
+ int idx, IgniteWriteAheadLogManager wal)
+ throws IgniteCheckedException {
+ unsetFlags(DIRTY);
+
+ RowLinkIO rowIo = (RowLinkIO)io;
+
+ // Check if entry is locked on primary node.
+ if (isFlagsSet(PRIMARY | FIRST)) {
+ long lockCrd = rowIo.getMvccLockCoordinatorVersion(pageAddr, idx);
+ long lockCntr = rowIo.getMvccLockCounter(pageAddr, idx);
+
+ // We cannot continue while entry is locked by another transaction.
+ if ((lockCrd != mvccCrd || lockCntr != mvccCntr)
+ && isActive(cctx, lockCrd, lockCntr, mvccSnapshot)) {
+ resCrd = lockCrd;
+ resCntr = lockCntr;
+
+ res = ResultType.LOCKED;
+
+ return setFlags(STOP);
+ }
+ }
+
+ MvccDataRow row = (MvccDataRow)tree.getRow(io, pageAddr, idx, RowData.LINK_WITH_HEADER);
+
+ // Check whether the row was updated by current transaction.
+ // In this case the row is already locked by current transaction and visible to it.
+ if (isFlagsSet(FIRST)) {
+ boolean removed = row.newMvccCoordinatorVersion() != MVCC_CRD_COUNTER_NA;
+
+ long rowCrd, rowCntr; int rowOpCntr;
+
+ if (removed) {
+ rowCrd = row.newMvccCoordinatorVersion();
+ rowCntr = row.newMvccCounter();
+ rowOpCntr = row.newMvccOperationCounter();
+ }
+ else {
+ rowCrd = row.mvccCoordinatorVersion();
+ rowCntr = row.mvccCounter();
+ rowOpCntr = row.mvccOperationCounter();
+ }
+
+ if (compare(mvccSnapshot, rowCrd, rowCntr) == 0) {
+ res = mvccOpCntr == rowOpCntr ? ResultType.VERSION_FOUND :
+ removed ? ResultType.PREV_NULL : ResultType.PREV_NOT_NULL;
+
+ if (removed)
+ setFlags(DELETED);
+ else
+ oldRow = row;
+
+ setFlags(LAST_COMMITTED_FOUND);
+ }
+ }
+
+ long rowLink = row.link();
+
+ long rowCrd = row.mvccCoordinatorVersion();
+ long rowCntr = row.mvccCounter();
+
+ // with hint bits
+ int rowOpCntr = (row.mvccTxState() << MVCC_HINTS_BIT_OFF) | (row.mvccOperationCounter() & ~MVCC_HINTS_MASK);
+
+ long rowNewCrd = row.newMvccCoordinatorVersion();
+ long rowNewCntr = row.newMvccCounter();
+
+ // with hint bits
+ int rowNewOpCntr = (row.newMvccTxState() << MVCC_HINTS_BIT_OFF) | (row.newMvccOperationCounter() & ~MVCC_HINTS_MASK);
+
+ // Search for youngest committed by another transaction row.
+ if (!isFlagsSet(LAST_COMMITTED_FOUND)) {
+ if (!(resCrd == rowCrd && resCntr == rowCntr)) { // It's possible it is a chain of aborted changes
+ byte txState = MvccUtils.state(cctx, rowCrd, rowCntr, rowOpCntr);
+
+ if (txState == TxState.COMMITTED) {
+ setFlags(LAST_COMMITTED_FOUND);
+
+ if (rowNewCrd != MVCC_CRD_COUNTER_NA) {
+ if (rowNewCrd == rowCrd && rowNewCntr == rowCntr)
+ // Row was deleted by the same Tx it was created
+ txState = TxState.COMMITTED;
+ else if (rowNewCrd == resCrd && rowNewCntr == resCntr)
+ // The row is linked to the previously checked aborted version;
+ txState = TxState.ABORTED;
+ else
+ // Check with TxLog if removed version is committed;
+ txState = MvccUtils.state(cctx, rowNewCrd, rowNewCntr, rowNewOpCntr);
+
+ if (!(txState == TxState.COMMITTED || txState == TxState.ABORTED))
+ throw unexpectedStateException(cctx, txState, rowNewCrd, rowNewCntr, rowNewOpCntr, mvccSnapshot);
+
+ if (txState == TxState.COMMITTED)
+ setFlags(DELETED);
+ }
+
+ if (isFlagsSet(DELETED))
+ res = ResultType.PREV_NULL;
+ else {
+ res = ResultType.PREV_NOT_NULL;
+
+ oldRow = row;
+ }
+
+ if (isFlagsSet(CHECK_VERSION)) {
+ long crdVer, cntr; int opCntr;
+
+ if (isFlagsSet(DELETED)) {
+ crdVer = rowNewCrd;
+ cntr = rowNewCntr;
+ opCntr = rowNewOpCntr;
+ }
+ else {
+ crdVer = rowCrd;
+ cntr = rowCntr;
+ opCntr = rowOpCntr;
+ }
+
+ // If last committed row is not visible it is possible write conflict.
+ if (!isVisible(cctx, mvccSnapshot, crdVer, cntr, opCntr, false)) {
+ // In case when row is accessed without previous version check (FAST_UPDATE)
+ // it is possible that we should consider this row non existent for current transaction
+ // without signalling write conflict.
+ // To do this we need to find youngest visible version and if it is removed version
+ // or there is no visible version then there is no conflict.
+ if (isFlagsSet(FAST_UPDATE)
+ && !(isFlagsSet(DELETED)
+ && isVisible(cctx, mvccSnapshot, rowCrd, rowCntr, rowOpCntr, false))) {
+ res = ResultType.PREV_NULL;
+
+ setFlags(FAST_MISMATCH);
+ }
+ else {
+ resCrd = crdVer;
+ resCntr = cntr;
+
+ res = ResultType.VERSION_MISMATCH; // Write conflict.
+
+ return setFlags(STOP);
+ }
+ }
+ }
+
+ // Lock entry for primary partition if needed.
+ // If invisible row is found for FAST_UPDATE case we should not lock row.
+ if (isFlagsSet(PRIMARY | REMOVE_OR_LOCK) && !isFlagsSet(FAST_MISMATCH)) {
+ rowIo.setMvccLockCoordinatorVersion(pageAddr, idx, mvccCrd);
+ rowIo.setMvccLockCounter(pageAddr, idx, mvccCntr);
+
+ // TODO Delta record IGNITE-7991
+
+ setFlags(DIRTY);
+ }
+
+ unsetFlags(CAN_WRITE); // No need to acquire write locks anymore
+ }
+ else if (txState == TxState.ABORTED) { // save aborted version to fast check new version of next row
+ resCrd = rowCrd;
+ resCntr = rowCntr;
+ }
+ else
+ throw unexpectedStateException(cctx, txState, rowCrd, rowCntr, rowOpCntr, mvccSnapshot);
+ }
+ }
+ // Search for youngest visible row.
+ // If we have not found any visible version then we does not see this row.
+ else if (isFlagsSet(FAST_MISMATCH)) {
+ assert !isFlagsSet(CAN_CLEANUP);
+ assert mvccVersionIsValid(rowNewCrd, rowNewCntr, rowNewOpCntr);
+
+ // Update version could be visible only if it is removal version,
+ // previous create versions were already checked in previous step and are definitely invisible.
+ // If we found visible removal version then we does not see this row.
+ if (isVisible(cctx, mvccSnapshot, rowNewCrd, rowNewCntr, rowNewOpCntr, false))
+ unsetFlags(FAST_MISMATCH);
+ // If the youngest visible for current transaction version is not removal version then it is write conflict.
+ else if (isVisible(cctx, mvccSnapshot, rowCrd, rowCntr, rowOpCntr, false)) {
+ resCrd = rowCrd;
+ resCntr = rowCntr;
+
+ res = ResultType.VERSION_MISMATCH;
+
+ return setFlags(STOP);
+ }
+ }
+
+ long cleanupVer = mvccSnapshot.cleanupVersion();
+
+ if (cleanupVer > MVCC_OP_COUNTER_NA // Do not clean if cleanup version is not assigned.
+ && !isFlagsSet(CAN_CLEANUP)
+ && isFlagsSet(LAST_COMMITTED_FOUND | DELETED)) {
+ assert mvccVersionIsValid(rowNewCrd, rowNewCntr, rowNewOpCntr);
+
+ // We can cleanup previous row only if it was deleted by another
+ // transaction and delete version is less or equal to cleanup one
+ if (rowNewCrd < mvccCrd || Long.compare(cleanupVer, rowNewCntr) >= 0)
+ setFlags(CAN_CLEANUP);
+ }
+
+ if (isFlagsSet(CAN_CLEANUP)
+ || !isFlagsSet(LAST_COMMITTED_FOUND)) { // can cleanup aborted versions
+ if (cleanupRows == null)
+ cleanupRows = new ArrayList<>();
+
+ cleanupRows.add(new MvccLinkAwareSearchRow(cacheId, key, rowCrd, rowCntr, rowOpCntr & ~MVCC_HINTS_MASK, rowLink));
+ }
+ else {
+ // Row obsoleted by current operation, all rows created or updated with current tx.
+ if (isFlagsSet(NEED_HISTORY)
+ && (row == oldRow
+ || (rowCrd == mvccCrd && rowCntr == mvccCntr)
+ || (rowNewCrd == mvccCrd && rowNewCntr == mvccCntr))) {
+ if (historyRows == null)
+ historyRows = new ArrayList<>();
+
+ historyRows.add(new MvccLinkAwareSearchRow(cacheId, key, rowCrd, rowCntr, rowOpCntr & ~MVCC_HINTS_MASK, rowLink));
+ }
+
+ if (cleanupVer > MVCC_OP_COUNTER_NA // Do not clean if cleanup version is not assigned.
+ && !isFlagsSet(CAN_CLEANUP)
+ && isFlagsSet(LAST_COMMITTED_FOUND)
+ && (rowCrd < mvccCrd || Long.compare(cleanupVer, rowCntr) >= 0))
+ // all further versions are guaranteed to be less than cleanup version
+ setFlags(CAN_CLEANUP);
+ }
+
+ return unsetFlags(FIRST);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int state() {
+ return state;
+ }
+
+ /**
+ * @return Old row.
+ */
+ public CacheDataRow oldRow() {
+ return oldRow;
+ }
+
+ /**
+ * @return {@code True} if previous value was non-null.
+ */
+ @Override public ResultType resultType() {
+ return res == null ? ResultType.PREV_NULL : res;
+ }
+
+ /**
+ * @return Rows which are safe to cleanup.
+ */
+ public List<MvccLinkAwareSearchRow> cleanupRows() {
+ return cleanupRows;
+ }
+
+ /**
+ * @return Result version.
+ */
+ @Override public MvccVersion resultVersion() {
+ switch (resultType()) {
+ case VERSION_FOUND:
+ case PREV_NULL:
+
+ return new MvccVersionImpl(mvccCrd, mvccCntr, mvccOpCntr);
+ case PREV_NOT_NULL:
+
+ return new MvccVersionImpl(oldRow.mvccCoordinatorVersion(), oldRow.mvccCounter(), oldRow.mvccOperationCounter());
+ case LOCKED:
+ case VERSION_MISMATCH:
+
+ assert resCrd != MVCC_CRD_COUNTER_NA && resCntr != MVCC_COUNTER_NA;
+
+ return new MvccVersionImpl(resCrd, resCntr, MVCC_OP_COUNTER_NA);
+ default:
+
+ throw new IllegalStateException("Unexpected result type: " + resultType());
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<MvccLinkAwareSearchRow> history() {
+ if (isFlagsSet(NEED_HISTORY) && historyRows == null)
+ historyRows = new ArrayList<>();
+
+ return historyRows;
+ }
+
+ /** */
+ private boolean isFlagsSet(int flags) {
+ return (state & flags) == flags;
+ }
+
+ /** */
+ private int setFlags(int flags) {
+ return state |= flags;
+ }
+
+ /** */
+ private int unsetFlags(int flags) {
+ return state &= (~flags);
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(MvccUpdateDataRow.class, this, "super", super.toString());
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccUpdateDataRowNative.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccUpdateDataRowNative.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccUpdateDataRowNative.java
new file mode 100644
index 0000000..38611a9
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccUpdateDataRowNative.java
@@ -0,0 +1,240 @@
+/*
+ * 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.tree.mvcc.data;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.processors.cache.CacheObject;
+import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccUtils;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccVersion;
+import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
+import org.apache.ignite.internal.processors.cache.persistence.CacheSearchRow;
+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.cache.tree.RowLinkIO;
+import org.apache.ignite.internal.processors.cache.tree.mvcc.search.MvccLinkAwareSearchRow;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.util.GridLongList;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ *
+ */
+public class MvccUpdateDataRowNative extends MvccDataRow implements BPlusTree.TreeRowClosure<CacheSearchRow, CacheDataRow> {
+ /** */
+ private final MvccSnapshot mvccSnapshot;
+ /** */
+ private ResultType res;
+ /** */
+ private boolean canCleanup;
+ /** */
+ private GridLongList activeTxs;
+ /** */
+ private List<MvccLinkAwareSearchRow> cleanupRows;
+ /** */
+ private CacheDataRow oldRow;
+
+ /**
+ * @param key Key.
+ * @param val Value.
+ * @param ver Version.
+ * @param expireTime Expire time.
+ * @param mvccSnapshot MVCC snapshot.
+ * @param newVer Update version.
+ * @param part Partition.
+ * @param cctx Cache context.
+ */
+ public MvccUpdateDataRowNative(
+ KeyCacheObject key,
+ CacheObject val,
+ GridCacheVersion ver,
+ long expireTime,
+ MvccSnapshot mvccSnapshot,
+ MvccVersion newVer,
+ int part,
+ GridCacheContext cctx) {
+ super(key,
+ val,
+ ver,
+ part,
+ expireTime,
+ cctx.cacheId(),
+ mvccSnapshot,
+ newVer);
+
+ this.mvccSnapshot = mvccSnapshot;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean apply(BPlusTree<CacheSearchRow, CacheDataRow> tree,
+ BPlusIO<CacheSearchRow> io,
+ long pageAddr,
+ int idx)
+ throws IgniteCheckedException {
+ RowLinkIO rowIo = (RowLinkIO)io;
+
+ // Assert version grows.
+ assert assertVersion(rowIo, pageAddr, idx);
+
+ boolean checkActive = mvccSnapshot.activeTransactions().size() > 0;
+
+ boolean txActive = false;
+
+ long rowCrdVer = rowIo.getMvccCoordinatorVersion(pageAddr, idx);
+
+ long crdVer = mvccCoordinatorVersion();
+
+ boolean isFirstRmvd = false;
+
+ if (res == null) {
+ int cmp = Long.compare(crdVer, rowCrdVer);
+
+ if (cmp == 0)
+ cmp = Long.compare(mvccSnapshot.counter(), rowIo.getMvccCounter(pageAddr, idx));
+
+ if (cmp == 0)
+ res = ResultType.VERSION_FOUND;
+ else {
+ oldRow = tree.getRow(io, pageAddr, idx, RowData.LINK_WITH_HEADER);
+
+ isFirstRmvd = oldRow.newMvccCoordinatorVersion() != 0;
+
+ if (isFirstRmvd)
+ res = ResultType.PREV_NULL;
+ else
+ res = ResultType.PREV_NOT_NULL;
+ }
+ }
+
+ // Suppose transactions on previous coordinator versions are done.
+ if (checkActive && crdVer == rowCrdVer) {
+ long rowMvccCntr = rowIo.getMvccCounter(pageAddr, idx);
+
+ long activeTx = isFirstRmvd ? oldRow.newMvccCounter() : rowMvccCntr;
+
+ if (mvccSnapshot.activeTransactions().contains(activeTx)) {
+ txActive = true;
+
+ if (activeTxs == null)
+ activeTxs = new GridLongList();
+
+ activeTxs.add(activeTx);
+ }
+ }
+
+ if (!txActive) {
+ assert Long.compare(crdVer, rowCrdVer) >= 0;
+
+ int cmp;
+
+ long rowCntr = rowIo.getMvccCounter(pageAddr, idx);
+
+ if (crdVer == rowCrdVer)
+ cmp = Long.compare(mvccSnapshot.cleanupVersion(), rowCntr);
+ else
+ cmp = 1;
+
+ if (cmp >= 0) {
+ // Do not cleanup oldest version.
+ if (canCleanup) {
+ assert MvccUtils.mvccVersionIsValid(rowCrdVer, rowCntr);
+
+ // Should not be possible to cleanup active tx.
+ assert rowCrdVer != crdVer || !mvccSnapshot.activeTransactions().contains(rowCntr);
+
+ if (cleanupRows == null)
+ cleanupRows = new ArrayList<>();
+
+ cleanupRows.add(new MvccLinkAwareSearchRow(cacheId, key, rowCrdVer, rowCntr,
+ rowIo.getMvccOperationCounter(pageAddr, idx), rowIo.getLink(pageAddr, idx)));
+ }
+ else
+ canCleanup = true;
+ }
+ }
+
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int mvccOperationCounter() {
+ return MvccUtils.MVCC_START_OP_CNTR;
+ }
+
+ /**
+ * @return Old row.
+ */
+ public CacheDataRow oldRow() {
+ return oldRow;
+ }
+
+ /**
+ * @return {@code True} if previous value was non-null.
+ */
+ public ResultType resultType() {
+ return res == null ? ResultType.PREV_NULL : res;
+ }
+
+ /**
+ * @return Active transactions to wait for.
+ */
+ @Nullable public GridLongList activeTransactions() {
+ return activeTxs;
+ }
+
+ /**
+ * @return Rows which are safe to cleanup.
+ */
+ public List<MvccLinkAwareSearchRow> cleanupRows() {
+ return cleanupRows;
+ }
+
+ /**
+ * @param io IO.
+ * @param pageAddr Page address.
+ * @param idx Item index.
+ * @return Always {@code true}.
+ */
+ private boolean assertVersion(RowLinkIO io, long pageAddr, int idx) {
+ long rowCrdVer = io.getMvccCoordinatorVersion(pageAddr, idx);
+ long rowCntr = io.getMvccCounter(pageAddr, idx);
+
+ int cmp = Long.compare(mvccCoordinatorVersion(), rowCrdVer);
+
+ if (cmp == 0)
+ cmp = Long.compare(mvccSnapshot.counter(), rowCntr);
+
+ // Can be equals if execute update on backup and backup already rebalanced value updated on primary.
+ assert cmp >= 0 : "[updCrd=" + mvccCoordinatorVersion() +
+ ", updCntr=" + mvccSnapshot.counter() +
+ ", rowCrd=" + rowCrdVer +
+ ", rowCntr=" + rowCntr + ']';
+
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(MvccUpdateDataRowNative.class, this, "super", super.toString());
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccUpdateResult.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccUpdateResult.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccUpdateResult.java
new file mode 100644
index 0000000..086a30d
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/MvccUpdateResult.java
@@ -0,0 +1,43 @@
+/*
+ * 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.tree.mvcc.data;
+
+import java.util.List;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccVersion;
+import org.apache.ignite.internal.processors.cache.tree.mvcc.search.MvccLinkAwareSearchRow;
+
+/**
+ *
+ */
+public interface MvccUpdateResult {
+ /**
+ * @return Type of result.
+ */
+ public ResultType resultType();
+
+ /**
+ * @return Result version.
+ */
+ public MvccVersion resultVersion();
+
+ /**
+ *
+ * @return Collection of row created or affected by the current tx.
+ */
+ public List<MvccLinkAwareSearchRow> history();
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/ResultType.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/ResultType.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/ResultType.java
new file mode 100644
index 0000000..eecb4a5
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/data/ResultType.java
@@ -0,0 +1,34 @@
+/*
+ * 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.tree.mvcc.data;
+
+/**
+ *
+ */
+public enum ResultType {
+ /** */
+ VERSION_FOUND,
+ /** */
+ PREV_NULL,
+ /** */
+ PREV_NOT_NULL,
+ /** */
+ LOCKED,
+ /** */
+ VERSION_MISMATCH
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccFirstRowTreeClosure.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccFirstRowTreeClosure.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccFirstRowTreeClosure.java
new file mode 100644
index 0000000..8546a66
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccFirstRowTreeClosure.java
@@ -0,0 +1,76 @@
+/*
+ * 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.tree.mvcc.search;
+
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccVersion;
+import org.apache.ignite.internal.processors.cache.mvcc.txlog.TxState;
+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.persistence.CacheSearchRow;
+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.cache.tree.RowLinkIO;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.jetbrains.annotations.Nullable;
+
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.getNewVersion;
+
+/**
+ * Closure which returns the very first encountered row.
+ */
+public class MvccFirstRowTreeClosure implements MvccTreeClosure {
+ /** */
+ private final GridCacheContext cctx;
+
+ /** */
+ private CacheDataRow res;
+
+ /**
+ * @param cctx Cache context.
+ */
+ public MvccFirstRowTreeClosure(GridCacheContext cctx) {
+ this.cctx = cctx;
+ }
+
+ /**
+ * @return Found row.
+ */
+ @Nullable public CacheDataRow row() {
+ return res;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean apply(BPlusTree<CacheSearchRow, CacheDataRow> tree, BPlusIO<CacheSearchRow> io,
+ long pageAddr, int idx) throws IgniteCheckedException {
+ RowLinkIO rowIo = (RowLinkIO)io;
+
+ MvccVersion newVersion = getNewVersion(cctx, rowIo.getLink(pageAddr, idx));
+
+ if (newVersion == null)
+ res = tree.getRow(io, pageAddr, idx, CacheDataRowAdapter.RowData.NO_KEY);
+
+ return false; // Stop search.
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(MvccFirstRowTreeClosure.class, this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccLinkAwareSearchRow.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccLinkAwareSearchRow.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccLinkAwareSearchRow.java
new file mode 100644
index 0000000..1e763a6
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccLinkAwareSearchRow.java
@@ -0,0 +1,48 @@
+/*
+ * 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.tree.mvcc.search;
+
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+
+/**
+ * MVCC search row which contains a link. Now used only for cleanup purposes.
+ */
+public class MvccLinkAwareSearchRow extends MvccSearchRow {
+ /** */
+ private final long link;
+
+ /**
+ * @param cacheId Cache ID.
+ * @param key Key.
+ * @param crdVer Mvcc coordinator version.
+ * @param mvccCntr Mvcc counter.
+ * @param link Link.
+ */
+ public MvccLinkAwareSearchRow(int cacheId, KeyCacheObject key, long crdVer, long mvccCntr, int mvccOpCntr, long link) {
+ super(cacheId, key, crdVer, mvccCntr, mvccOpCntr);
+
+ assert link != 0L;
+
+ this.link = link;
+ }
+
+ /** {@inheritDoc} */
+ @Override public long link() {
+ return link;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccMaxSearchRow.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccMaxSearchRow.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccMaxSearchRow.java
new file mode 100644
index 0000000..f5f3b67
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccMaxSearchRow.java
@@ -0,0 +1,55 @@
+/*
+ * 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.tree.mvcc.search;
+
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.tree.SearchRow;
+import org.apache.ignite.internal.util.typedef.internal.S;
+
+/**
+ * Search row for maximum key version.
+ */
+public class MvccMaxSearchRow extends SearchRow {
+ /**
+ * @param cacheId Cache ID.
+ * @param key Key.
+ */
+ public MvccMaxSearchRow(int cacheId, KeyCacheObject key) {
+ super(cacheId, key);
+ }
+
+ /** {@inheritDoc} */
+ @Override public long mvccCoordinatorVersion() {
+ return Long.MAX_VALUE;
+ }
+
+ /** {@inheritDoc} */
+ @Override public long mvccCounter() {
+ return Long.MAX_VALUE;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int mvccOperationCounter() {
+ return Integer.MAX_VALUE;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(MvccMaxSearchRow.class, this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccMinSearchRow.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccMinSearchRow.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccMinSearchRow.java
new file mode 100644
index 0000000..a1cb8fc
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccMinSearchRow.java
@@ -0,0 +1,55 @@
+/*
+ * 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.tree.mvcc.search;
+
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.tree.SearchRow;
+import org.apache.ignite.internal.util.typedef.internal.S;
+
+/**
+ * Search row for minimum key version.
+ */
+public class MvccMinSearchRow extends SearchRow {
+ /**
+ * @param cacheId Cache ID.
+ * @param key Key.
+ */
+ public MvccMinSearchRow(int cacheId, KeyCacheObject key) {
+ super(cacheId, key);
+ }
+
+ /** {@inheritDoc} */
+ @Override public long mvccCoordinatorVersion() {
+ return 1L;
+ }
+
+ /** {@inheritDoc} */
+ @Override public long mvccCounter() {
+ return 1L;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int mvccOperationCounter() {
+ return 1;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(MvccMinSearchRow.class, this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccSearchRow.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccSearchRow.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccSearchRow.java
new file mode 100644
index 0000000..d0510f7
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccSearchRow.java
@@ -0,0 +1,70 @@
+/*
+ * 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.tree.mvcc.search;
+
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.tree.SearchRow;
+import org.apache.ignite.internal.util.typedef.internal.S;
+
+/**
+ * MVCC search row which uses specific MVCC version (coordinator + counter) to filter rows.
+ */
+public class MvccSearchRow extends SearchRow {
+ /** Coordinator version. */
+ protected long crdVer;
+
+ /** Counter. */
+ private long cntr;
+
+ /** Operation counter. */
+ protected int opCntr;
+
+ /**
+ * @param cacheId Cache ID.
+ * @param key Key.
+ * @param crdVer Coordinator version.
+ * @param cntr Counter.
+ */
+ public MvccSearchRow(int cacheId, KeyCacheObject key, long crdVer, long cntr, int opCntr) {
+ super(cacheId, key);
+
+ this.crdVer = crdVer;
+ this.cntr = cntr;
+ this.opCntr = opCntr;
+ }
+
+ /** {@inheritDoc} */
+ @Override public long mvccCoordinatorVersion() {
+ return crdVer;
+ }
+
+ /** {@inheritDoc} */
+ @Override public long mvccCounter() {
+ return cntr;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int mvccOperationCounter() {
+ return opCntr;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(MvccSearchRow.class, this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccSnapshotSearchRow.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccSnapshotSearchRow.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccSnapshotSearchRow.java
new file mode 100644
index 0000000..432593d
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccSnapshotSearchRow.java
@@ -0,0 +1,113 @@
+/*
+ * 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.tree.mvcc.search;
+
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccUtils;
+import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
+import org.apache.ignite.internal.processors.cache.persistence.CacheDataRowAdapter;
+import org.apache.ignite.internal.processors.cache.persistence.CacheSearchRow;
+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.cache.tree.RowLinkIO;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.jetbrains.annotations.Nullable;
+
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_INVISIBLE;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_VISIBLE;
+import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.MVCC_VISIBLE_REMOVED;
+
+/**
+ * Search row which returns the first row visible for the given snapshot. Usage:
+ * - set this row as the upper bound
+ * - pass the same row as search closure.
+ */
+public class MvccSnapshotSearchRow extends MvccSearchRow implements MvccTreeClosure {
+ /** */
+ private final GridCacheContext cctx;
+
+ /** Resulting row. */
+ private CacheDataRow res;
+
+ /** */
+ private MvccSnapshot snapshot;
+
+ /**
+ * Constructor.
+ *
+ * @param cctx
+ * @param key Key.
+ * @param snapshot Snapshot.
+ */
+ public MvccSnapshotSearchRow(GridCacheContext cctx, KeyCacheObject key,
+ MvccSnapshot snapshot) {
+ super(cctx.cacheId(), key, snapshot.coordinatorVersion(), snapshot.counter(), Integer.MAX_VALUE);
+
+ this.cctx = cctx;
+
+ this.snapshot = snapshot;
+ }
+
+ /**
+ * @return Found row.
+ */
+ @Nullable public CacheDataRow row() {
+ return res;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean apply(BPlusTree<CacheSearchRow, CacheDataRow> tree, BPlusIO<CacheSearchRow> io,
+ long pageAddr, int idx) throws IgniteCheckedException {
+ RowLinkIO rowIo = (RowLinkIO)io;
+
+ long rowCrdVer = rowIo.getMvccCoordinatorVersion(pageAddr, idx);
+ long rowCntr = rowIo.getMvccCounter(pageAddr, idx);
+ int rowOpCntr = rowIo.getMvccOperationCounter(pageAddr, idx);
+
+ assert MvccUtils.mvccVersionIsValid(rowCrdVer, rowCntr, rowOpCntr);
+
+ if (MvccUtils.isVisible(cctx, snapshot, rowCrdVer, rowCntr, rowOpCntr, false)) {
+ int state = MvccUtils.getVisibleState(cctx, rowIo.getLink(pageAddr, idx), snapshot);
+
+ if (state == MVCC_INVISIBLE)
+ return true;
+
+ if (state == MVCC_VISIBLE_REMOVED)
+ res = null;
+ else {
+ assert state == MVCC_VISIBLE;
+
+ res = tree.getRow(io, pageAddr, idx, CacheDataRowAdapter.RowData.NO_KEY);
+
+ res.key(key());
+ }
+
+ return false; // Stop search.
+ }
+
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(MvccSnapshotSearchRow.class, this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccTreeClosure.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccTreeClosure.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccTreeClosure.java
new file mode 100644
index 0000000..031903f
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/mvcc/search/MvccTreeClosure.java
@@ -0,0 +1,29 @@
+/*
+ * 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.tree.mvcc.search;
+
+import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
+import org.apache.ignite.internal.processors.cache.persistence.CacheSearchRow;
+import org.apache.ignite.internal.processors.cache.persistence.tree.BPlusTree;
+
+/**
+ * Common interface for MVCC search row closures (no-op, only to have clear inheritance hierarchy).
+ */
+public interface MvccTreeClosure extends BPlusTree.TreeRowClosure<CacheSearchRow, CacheDataRow> {
+ // No-op.
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/SqlStateCode.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/SqlStateCode.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/SqlStateCode.java
index eff680f..68ac200 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/SqlStateCode.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/SqlStateCode.java
@@ -61,6 +61,9 @@ public final class SqlStateCode {
/** Requested operation is not supported. */
public final static String UNSUPPORTED_OPERATION = "0A000";
+ /** Transaction state exception. */
+ public final static String TRANSACTION_STATE_EXCEPTION = "25000";
+
/** Parsing exception. */
public final static String PARSING_EXCEPTION = "42000";
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/jdbc/JdbcBatchExecuteRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/jdbc/JdbcBatchExecuteRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/jdbc/JdbcBatchExecuteRequest.java
index bdc558c..065a298 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/jdbc/JdbcBatchExecuteRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/jdbc/JdbcBatchExecuteRequest.java
@@ -39,6 +39,9 @@ public class JdbcBatchExecuteRequest extends JdbcRequest {
@GridToStringInclude(sensitive = true)
private List<JdbcQuery> queries;
+ /** Client auto commit flag state. */
+ private boolean autoCommit;
+
/**
* Last stream batch flag - whether open streamers on current connection
* must be flushed and closed after this batch.
@@ -63,15 +66,17 @@ public class JdbcBatchExecuteRequest extends JdbcRequest {
/**
* @param schemaName Schema name.
* @param queries Queries.
+ * @param autoCommit Client auto commit flag state.
* @param lastStreamBatch {@code true} in case the request is the last batch at the stream.
*/
- public JdbcBatchExecuteRequest(String schemaName, List<JdbcQuery> queries, boolean lastStreamBatch) {
+ public JdbcBatchExecuteRequest(String schemaName, List<JdbcQuery> queries, boolean autoCommit, boolean lastStreamBatch) {
super(BATCH_EXEC);
assert lastStreamBatch || !F.isEmpty(queries);
this.schemaName = schemaName;
this.queries = queries;
+ this.autoCommit = autoCommit;
this.lastStreamBatch = lastStreamBatch;
}
@@ -81,15 +86,17 @@ public class JdbcBatchExecuteRequest extends JdbcRequest {
* @param type Request type.
* @param schemaName Schema name.
* @param queries Queries.
+ * @param autoCommit Client auto commit flag state.
* @param lastStreamBatch {@code true} in case the request is the last batch at the stream.
*/
- protected JdbcBatchExecuteRequest(byte type, String schemaName, List<JdbcQuery> queries, boolean lastStreamBatch) {
+ protected JdbcBatchExecuteRequest(byte type, String schemaName, List<JdbcQuery> queries, boolean autoCommit, boolean lastStreamBatch) {
super(type);
assert lastStreamBatch || !F.isEmpty(queries);
this.schemaName = schemaName;
this.queries = queries;
+ this.autoCommit = autoCommit;
this.lastStreamBatch = lastStreamBatch;
}
@@ -108,6 +115,13 @@ public class JdbcBatchExecuteRequest extends JdbcRequest {
}
/**
+ * @return Auto commit flag.
+ */
+ boolean autoCommit() {
+ return autoCommit;
+ }
+
+ /**
* @return Last stream batch flag.
*/
public boolean isLastStreamBatch() {
@@ -125,14 +139,17 @@ public class JdbcBatchExecuteRequest extends JdbcRequest {
for (JdbcQuery q : queries)
q.writeBinary(writer);
+
}
else
writer.writeInt(0);
+ writer.writeBoolean(autoCommit);
writer.writeBoolean(lastStreamBatch);
}
/** {@inheritDoc} */
+ @SuppressWarnings("SimplifiableIfStatement")
@Override public void readBinary(BinaryReaderExImpl reader) throws BinaryObjectException {
super.readBinary(reader);
@@ -151,8 +168,12 @@ public class JdbcBatchExecuteRequest extends JdbcRequest {
}
try {
- if (reader.available() > 0)
+ if (reader.available() > 0) {
+ autoCommit = reader.readBoolean();
lastStreamBatch = reader.readBoolean();
+ }
+ else
+ autoCommit = true;
}
catch (IOException e) {
throw new BinaryObjectException(e);
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/jdbc/JdbcConnectionContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/jdbc/JdbcConnectionContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/jdbc/JdbcConnectionContext.java
index 23f4d49..9f3fe2a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/jdbc/JdbcConnectionContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/jdbc/JdbcConnectionContext.java
@@ -29,8 +29,10 @@ import org.apache.ignite.internal.processors.odbc.ClientListenerMessageParser;
import org.apache.ignite.internal.processors.odbc.ClientListenerProtocolVersion;
import org.apache.ignite.internal.processors.odbc.ClientListenerRequestHandler;
import org.apache.ignite.internal.processors.odbc.ClientListenerResponse;
+import org.apache.ignite.internal.processors.query.NestedTxMode;
import org.apache.ignite.internal.util.GridSpinBusyLock;
import org.apache.ignite.internal.util.nio.GridNioSession;
+import org.apache.ignite.internal.util.typedef.F;
/**
* JDBC Connection Context.
@@ -129,29 +131,44 @@ public class JdbcConnectionContext extends ClientListenerAbstractConnectionConte
boolean autoCloseCursors = reader.readBoolean();
boolean lazyExec = false;
+ boolean skipReducerOnUpdate = false;
+
+ NestedTxMode nestedTxMode = NestedTxMode.DEFAULT;
+ AuthorizationContext actx = null;
if (ver.compareTo(VER_2_1_5) >= 0)
lazyExec = reader.readBoolean();
- boolean skipReducerOnUpdate = false;
-
if (ver.compareTo(VER_2_3_0) >= 0)
skipReducerOnUpdate = reader.readBoolean();
- String user = null;
- String passwd = null;
+ if (ver.compareTo(VER_2_5_0) >= 0) {
+ String nestedTxModeName = reader.readString();
- try {
- if (reader.available() > 0) {
- user = reader.readString();
- passwd = reader.readString();
+ if (!F.isEmpty(nestedTxModeName)) {
+ try {
+ nestedTxMode = NestedTxMode.valueOf(nestedTxModeName);
+ }
+ catch (IllegalArgumentException e) {
+ throw new IgniteCheckedException("Invalid nested transactions handling mode: " + nestedTxModeName);
+ }
}
- }
- catch (Exception e) {
- throw new IgniteCheckedException("Handshake error: " + e.getMessage(), e);
- }
- AuthorizationContext actx = authenticate(user, passwd);
+ String user = null;
+ String passwd = null;
+
+ try {
+ if (reader.available() > 0) {
+ user = reader.readString();
+ passwd = reader.readString();
+ }
+ }
+ catch (Exception e) {
+ throw new IgniteCheckedException("Handshake error: " + e.getMessage(), e);
+ }
+
+ actx = authenticate(user, passwd);
+ }
parser = new JdbcMessageParser(ctx);
@@ -169,7 +186,9 @@ public class JdbcConnectionContext extends ClientListenerAbstractConnectionConte
};
handler = new JdbcRequestHandler(ctx, busyLock, sender, maxCursors, distributedJoins, enforceJoinOrder,
- collocated, replicatedOnly, autoCloseCursors, lazyExec, skipReducerOnUpdate, actx, ver);
+ collocated, replicatedOnly, autoCloseCursors, lazyExec, skipReducerOnUpdate, nestedTxMode, actx, ver);
+
+ handler.start();
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/jdbc/JdbcOrderedBatchExecuteRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/jdbc/JdbcOrderedBatchExecuteRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/jdbc/JdbcOrderedBatchExecuteRequest.java
index 3e84731..1d4d4ac 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/jdbc/JdbcOrderedBatchExecuteRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/jdbc/JdbcOrderedBatchExecuteRequest.java
@@ -42,12 +42,13 @@ public class JdbcOrderedBatchExecuteRequest extends JdbcBatchExecuteRequest
/**
* @param schemaName Schema name.
* @param queries Queries.
+ * @param autoCommit Client auto commit flag state.
* @param lastStreamBatch {@code true} in case the request is the last batch at the stream.
* @param order Request order.
*/
public JdbcOrderedBatchExecuteRequest(String schemaName, List<JdbcQuery> queries,
- boolean lastStreamBatch, long order) {
- super(BATCH_EXEC_ORDERED, schemaName, queries, lastStreamBatch);
+ boolean autoCommit, boolean lastStreamBatch, long order) {
+ super(BATCH_EXEC_ORDERED, schemaName, queries, autoCommit, lastStreamBatch);
this.order = order;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/00bb4d4d/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/jdbc/JdbcQueryExecuteRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/jdbc/JdbcQueryExecuteRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/jdbc/JdbcQueryExecuteRequest.java
index 3e54fc8..c13117a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/jdbc/JdbcQueryExecuteRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/jdbc/JdbcQueryExecuteRequest.java
@@ -51,6 +51,9 @@ public class JdbcQueryExecuteRequest extends JdbcRequest {
/** Expected statement type. */
private JdbcStatementType stmtType;
+ /** Client auto commit flag state. */
+ private boolean autoCommit;
+
/**
*/
JdbcQueryExecuteRequest() {
@@ -62,11 +65,12 @@ public class JdbcQueryExecuteRequest extends JdbcRequest {
* @param schemaName Cache name.
* @param pageSize Fetch size.
* @param maxRows Max rows.
+ * @param autoCommit Connection auto commit flag state.
* @param sqlQry SQL query.
* @param args Arguments list.
*/
public JdbcQueryExecuteRequest(JdbcStatementType stmtType, String schemaName, int pageSize, int maxRows,
- String sqlQry, Object[] args) {
+ boolean autoCommit, String sqlQry, Object[] args) {
super(QRY_EXEC);
this.schemaName = F.isEmpty(schemaName) ? null : schemaName;
@@ -75,6 +79,7 @@ public class JdbcQueryExecuteRequest extends JdbcRequest {
this.sqlQry = sqlQry;
this.args = args;
this.stmtType = stmtType;
+ this.autoCommit = autoCommit;
}
/**
@@ -119,6 +124,13 @@ public class JdbcQueryExecuteRequest extends JdbcRequest {
return stmtType;
}
+ /**
+ * @return Auto commit flag.
+ */
+ boolean autoCommit() {
+ return autoCommit;
+ }
+
/** {@inheritDoc} */
@Override public void writeBinary(BinaryWriterExImpl writer) throws BinaryObjectException {
super.writeBinary(writer);
@@ -136,9 +148,12 @@ public class JdbcQueryExecuteRequest extends JdbcRequest {
}
writer.writeByte((byte)stmtType.ordinal());
+
+ writer.writeBoolean(autoCommit);
}
/** {@inheritDoc} */
+ @SuppressWarnings("SimplifiableIfStatement")
@Override public void readBinary(BinaryReaderExImpl reader) throws BinaryObjectException {
super.readBinary(reader);
@@ -163,6 +178,16 @@ public class JdbcQueryExecuteRequest extends JdbcRequest {
catch (IOException e) {
throw new BinaryObjectException(e);
}
+
+ try {
+ if (reader.available() > 0)
+ autoCommit = reader.readBoolean();
+ else
+ autoCommit = true;
+ }
+ catch (IOException e) {
+ throw new BinaryObjectException(e);
+ }
}
/** {@inheritDoc} */