Merge pull request #954 from katzyn/undo_log

Avoid incorrect reads in iterators of TransactionMap

h2/src/main/org/h2/command/dml/Merge.java

@@ -118,6 +118,7 @@ public class Merge extends Prepared {
             }
         } else {
             // process select data for list
+            query.setNeverLazy(true);
             ResultInterface rows = query.query(0);
             count = 0;
             targetTable.fire(session, Trigger.UPDATE | Trigger.INSERT, true);

h2/src/main/org/h2/mvstore/db/TransactionStore.java

@@ -344,7 +344,7 @@ public class TransactionStore {
         }
         // TODO could synchronize on blocks (100 at a time or so)
         rwLock.writeLock().lock();
-        int oldStatus = t.status;
+        int oldStatus = t.getStatus();
         try {
             t.setStatus(Transaction.STATUS_COMMITTING);
             for (long logId = 0; logId < maxLogId; logId++) {
@@ -372,9 +372,7 @@ public class TransactionStore {
                             if (value.value == null) {
                                 map.remove(key);
                             } else {
-                                VersionedValue v2 = new VersionedValue();
-                                v2.value = value.value;
-                                map.put(key, v2);
+                                map.put(key, new VersionedValue(0L, value.value));
                             }
                         }
                     }
@@ -968,9 +966,10 @@ public class TransactionStore {
                     long size = 0;
                     Cursor<K, VersionedValue> cursor = map.cursor(null);
                     while (cursor.hasNext()) {
-                        VersionedValue data;
                         K key = cursor.next();
-                        data = getValue(key, readLogId, cursor.getValue());
+                        // cursor.getValue() returns outdated value
+                        VersionedValue data = map.get(key);
+                        data = getValue(key, readLogId, data);
                         if (data != null && data.value != null) {
                             size++;
                         }
@@ -1053,8 +1052,7 @@ public class TransactionStore {
         @SuppressWarnings("unchecked")
         public V putCommitted(K key, V value) {
             DataUtils.checkArgument(value != null, "The value may not be null");
-            VersionedValue newValue = new VersionedValue();
-            newValue.value = value;
+            VersionedValue newValue = new VersionedValue(0L, value);
             VersionedValue oldValue = map.put(key, newValue);
             return (V) (oldValue == null ? null : oldValue.value);
         }
@@ -1133,10 +1131,9 @@ public class TransactionStore {
                     }
                 }
             }
-            VersionedValue newValue = new VersionedValue();
-            newValue.operationId = getOperationId(
-                    transaction.transactionId, transaction.logId);
-            newValue.value = value;
+            VersionedValue newValue = new VersionedValue(
+                    getOperationId(transaction.transactionId, transaction.logId),
+                    value);
             if (current == null) {
                 // a new value
                 transaction.log(mapId, key, current);
@@ -1533,7 +1530,8 @@ public class TransactionStore {
                             if (to != null && map.getKeyType().compare(k, to) > 0) {
                                 break;
                             }
-                            VersionedValue data = cursor.getValue();
+                            // cursor.getValue() returns outdated value
+                            VersionedValue data = map.get(key);
                             data = getValue(key, readLogId, data);
                             if (data != null && data.value != null) {
                                 @SuppressWarnings("unchecked")
@@ -1641,12 +1639,17 @@ public class TransactionStore {
         /**
          * The operation id.
          */
-        public long operationId;
+        final long operationId;
 
         /**
          * The value.
          */
-        public Object value;
+        final Object value;
+
+        VersionedValue(long operationId, Object value) {
+            this.operationId = operationId;
+            this.value = value;
+        }
 
         @Override
         public String toString() {
@@ -1694,9 +1697,7 @@ public class TransactionStore {
             if (buff.get() == 0) {
                 // fast path (no op ids or null entries)
                 for (int i = 0; i < len; i++) {
-                    VersionedValue v = new VersionedValue();
-                    v.value = valueType.read(buff);
-                    obj[i] = v;
+                    obj[i] = new VersionedValue(0L, valueType.read(buff));
                 }
             } else {
                 // slow path (some entries may be null)
@@ -1708,12 +1709,14 @@ public class TransactionStore {
 
         @Override
         public Object read(ByteBuffer buff) {
-            VersionedValue v = new VersionedValue();
-            v.operationId = DataUtils.readVarLong(buff);
+            long operationId = DataUtils.readVarLong(buff);
+            Object value;
             if (buff.get() == 1) {
-                v.value = valueType.read(buff);
+                value = valueType.read(buff);
+            } else {
+                value = null;
             }
-            return v;
+            return new VersionedValue(operationId, value);
         }
 
         @Override

h2/src/test/org/h2/test/mvcc/TestMvccMultiThreaded2.java

@@ -12,6 +12,7 @@ import java.sql.SQLException;
 import java.sql.Statement;
 import java.util.ArrayList;
 import org.h2.jdbc.JdbcSQLException;
+import org.h2.message.DbException;
 import org.h2.test.TestBase;
 import org.h2.util.IOUtils;
 
@@ -63,6 +64,9 @@ public class TestMvccMultiThreaded2 extends TestBase {
         ps.setInt(1, 1);
         ps.setInt(2, 100);
         ps.executeUpdate();
+        ps.setInt(1, 2);
+        ps.setInt(2, 200);
+        ps.executeUpdate();
         conn.commit();
 
         ArrayList<SelectForUpdate> threads = new ArrayList<>();
@@ -80,9 +84,11 @@ public class TestMvccMultiThreaded2 extends TestBase {
         @SuppressWarnings("unused")
         int minProcessed = Integer.MAX_VALUE, maxProcessed = 0, totalProcessed = 0;
 
+        boolean allOk = true;
         for (SelectForUpdate sfu : threads) {
             // make sure all threads have stopped by joining with them
             sfu.join();
+            allOk &= sfu.ok;
             totalProcessed += sfu.iterationsProcessed;
             if (sfu.iterationsProcessed > maxProcessed) {
                 maxProcessed = sfu.iterationsProcessed;
@@ -102,6 +108,8 @@ public class TestMvccMultiThreaded2 extends TestBase {
 
         IOUtils.closeSilently(conn);
         deleteDb(getTestName());
+
+        assertTrue(allOk);
     }
 
     /**
@@ -111,6 +119,8 @@ public class TestMvccMultiThreaded2 extends TestBase {
 
         public int iterationsProcessed;
 
+        public boolean ok;
+
         SelectForUpdate() {
         }
 
@@ -130,11 +140,20 @@ public class TestMvccMultiThreaded2 extends TestBase {
                     try {
                         PreparedStatement ps = conn.prepareStatement(
                                 "SELECT * FROM test WHERE entity_id = ? FOR UPDATE");
-                        ps.setString(1, "1");
+                        String id;
+                        int value;
+                        if ((iterationsProcessed & 1) == 0) {
+                            id = "1";
+                            value = 100;
+                        } else {
+                            id = "2";
+                            value = 200;
+                        }
+                        ps.setString(1, id);
                         ResultSet rs = ps.executeQuery();
 
                         assertTrue(rs.next());
-                        assertTrue(rs.getInt(2) == 100);
+                        assertTrue(rs.getInt(2) == value);
 
                         conn.commit();
                         iterationsProcessed++;
@@ -149,11 +168,13 @@ public class TestMvccMultiThreaded2 extends TestBase {
                 }
             } catch (SQLException e) {
                 TestBase.logError("SQL error from thread "+getName(), e);
+                throw DbException.convert(e);
             } catch (Exception e) {
                 TestBase.logError("General error from thread "+getName(), e);
                 throw e;
             }
             IOUtils.closeSilently(conn);
+            ok = true;
         }
     }
 }