Push row locking down to Table implementation

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

@@ -614,7 +614,8 @@ public class Select extends Query {
             }
         }
         if (isForUpdateMvcc) {
-            topTableFilter.lockRows(forUpdateRows);
+            assert forUpdateRows != null;
+            topTableFilter.lockRows(forUpdateRows.iterator());
         }
         return null;
     }

h2/src/main/org/h2/table/Table.java

@@ -9,6 +9,7 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArrayList;
 
@@ -177,6 +178,23 @@ public abstract class Table extends SchemaObjectBase {
      */
     public abstract void removeRow(Session session, Row row);
 
+    /**
+     * Locks rows, preventing any updated to them, except from the session specified.
+     *
+     * @param session the session
+     * @param rowsForUpdate rows to lock
+     */
+    public void lockRows(Session session, Iterator<Row> rowsForUpdate) {
+        while (rowsForUpdate.hasNext()) {
+            Row row = rowsForUpdate.next();
+            Row newRow = row.getCopy();
+            removeRow(session, row);
+            session.log(this, UndoLogRecord.DELETE, row);
+            addRow(session, newRow);
+            session.log(this, UndoLogRecord.INSERT, newRow);
+        }
+    }
+
     /**
      * Remove all rows from the table and indexes.
      *

h2/src/main/org/h2/table/TableFilter.java

@@ -7,6 +7,7 @@ package org.h2.table;
 
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.Iterator;
 
 import org.h2.api.ErrorCode;
 import org.h2.command.Parser;
@@ -1158,14 +1159,8 @@ public class TableFilter implements ColumnResolver {
      *
      * @param forUpdateRows the rows to lock
      */
-    public void lockRows(ArrayList<Row> forUpdateRows) {
-        for (Row row : forUpdateRows) {
-            Row newRow = row.getCopy();
-            table.removeRow(session, row);
-            session.log(table, UndoLogRecord.DELETE, row);
-            table.addRow(session, newRow);
-            session.log(table, UndoLogRecord.INSERT, newRow);
-        }
+    public void lockRows(Iterator<Row> forUpdateRows) {
+        table.lockRows(session, forUpdateRows);
     }
 
     public TableFilter getNestedJoin() {